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()