Spaces:
Running
Running
| import asyncio | |
| import tempfile | |
| import unittest | |
| from pathlib import Path | |
| from unittest.mock import patch | |
| import server | |
| from fastapi.testclient import TestClient | |
| class ServerSecurityTests(unittest.TestCase): | |
| def setUp(self): | |
| server._runner_start_windows.clear() | |
| def test_only_absolute_https_links_are_exposed(self): | |
| self.assertEqual(server._safe_external_url("https://arcade.makecode.com/_abc"), "https://arcade.makecode.com/_abc") | |
| self.assertEqual(server._safe_external_url("javascript:alert(1)"), "") | |
| self.assertEqual(server._safe_external_url("https://user:pass@example.com/"), "") | |
| self.assertEqual(server._safe_external_url("http://example.com/"), "") | |
| def test_public_runner_starts_are_limited_per_ip(self): | |
| old_limit, old_cooldown = server.RUNNER_STARTS_PER_IP_PER_HOUR, server.RUNNER_START_COOLDOWN_SEC | |
| try: | |
| server.RUNNER_STARTS_PER_IP_PER_HOUR = 2 | |
| server.RUNNER_START_COOLDOWN_SEC = 0.5 | |
| with patch("server.time.monotonic", side_effect=[100, 101, 102, 103]): | |
| self.assertTrue(server.allow_runner_start("198.51.100.8")) | |
| self.assertTrue(server.allow_runner_start("198.51.100.8")) | |
| self.assertFalse(server.allow_runner_start("198.51.100.8")) | |
| self.assertTrue(server.allow_runner_start("198.51.100.9")) | |
| finally: | |
| server.RUNNER_STARTS_PER_IP_PER_HOUR = old_limit | |
| server.RUNNER_START_COOLDOWN_SEC = old_cooldown | |
| def test_runner_staging_skips_credentials_but_keeps_unrelated_named_files(self): | |
| with tempfile.TemporaryDirectory() as source_temp, tempfile.TemporaryDirectory() as scratch_temp: | |
| source = Path(source_temp) | |
| (source / "main.py").write_text("print('ok')\n", encoding="utf-8") | |
| (source / ".env").write_text("HF_TOKEN=should-not-copy\n", encoding="utf-8") | |
| (source / "credentials.json").write_text("{}", encoding="utf-8") | |
| (source / "secret_notes.py").write_text("NOTE = 'ordinary project file'\n", encoding="utf-8") | |
| with patch.object(server, "RUNNER_TMP_DIR", scratch_temp): | |
| session_root, workspace = server._stage_runner_project(str(source)) | |
| try: | |
| self.assertTrue((Path(workspace) / "main.py").is_file()) | |
| self.assertTrue((Path(workspace) / "secret_notes.py").is_file()) | |
| self.assertFalse((Path(workspace) / ".env").exists()) | |
| self.assertFalse((Path(workspace) / "credentials.json").exists()) | |
| finally: | |
| import shutil | |
| shutil.rmtree(session_root, ignore_errors=True) | |
| def test_sandbox_refuses_non_linux_execution(self): | |
| with patch.object(server.sys, "platform", "darwin"): | |
| with self.assertRaisesRegex(RuntimeError, "拒絕未隔離執行"): | |
| server._runner_sandbox_command("/tmp/project", "/tmp/session/runner.sock") | |
| def test_runner_stderr_capture_is_bounded(self): | |
| async def capture(): | |
| stream = asyncio.StreamReader() | |
| captured = bytearray() | |
| stream.feed_data(b"x" * 5000) | |
| stream.feed_eof() | |
| await server._capture_bounded_stderr(stream, captured) | |
| return captured | |
| self.assertEqual(len(asyncio.run(capture())), 4096) | |
| def test_explicit_process_mode_keeps_limits_and_uses_private_socket(self): | |
| with patch.object(server, "RUNNER_ISOLATION_MODE", "process"), \ | |
| patch.object(server.sys, "platform", "linux"), \ | |
| patch.object(server, "PRLIMIT_PATH", "/usr/bin/prlimit"): | |
| command = server._runner_command("/tmp/project", "/tmp/session/runner.sock") | |
| self.assertEqual(command[0], "/usr/bin/prlimit") | |
| self.assertIn("--nproc=64:64", command) | |
| self.assertIn("/tmp/project/main.py", command) | |
| self.assertIn("/tmp/session/runner.sock", command) | |
| self.assertNotIn("--unshare-user", command) | |
| def test_runner_environment_does_not_inherit_application_secrets(self): | |
| with patch.dict(server.os.environ, {"HF_TOKEN": "private", "ADMIN_PASSWORD": "private"}): | |
| environment = server._runner_environment("/tmp/project") | |
| self.assertNotIn("HF_TOKEN", environment) | |
| self.assertNotIn("ADMIN_PASSWORD", environment) | |
| self.assertEqual(environment["SDL_VIDEODRIVER"], "dummy") | |
| def test_invalid_mode_cannot_silently_disable_isolation(self): | |
| with patch.object(server, "RUNNER_ISOLATION_MODE", "typo"): | |
| with self.assertRaises(RuntimeError): | |
| server._runner_command("/tmp/project", "/tmp/session/runner.sock") | |
| def test_homepage_renders_template(self): | |
| with TestClient(server.app) as client: | |
| response = client.get("/") | |
| self.assertEqual(response.status_code, 200) | |
| self.assertIn("text/html", response.headers["content-type"]) | |
| self.assertIn( | |
| "frame-ancestors 'self' https://huggingface.co", | |
| response.headers["content-security-policy"], | |
| ) | |
| if __name__ == "__main__": | |
| unittest.main() | |