Spaces:
Running
Running
File size: 5,150 Bytes
81f4613 7547388 438ac1c 7547388 81f4613 4415019 438ac1c c939e29 438ac1c 7547388 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 | 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()
|