fix: Host/Origin check, reload guard, gear flicker, test isolation, toolbelt wait (#4, I-3, I-4, M-2, M-8, M-12)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

files/claude_agents.py

@@ -995,8 +995,9 @@ def render_page() -> str:
     try {{
       const r = await fetch('/limits', {{cache: 'no-store'}});
       if (r.ok) {{
+        const html = await r.text();
         document.getElementById('gear-home').appendChild(gear);  // иначе innerHTML уничтожит ⚙
-        document.getElementById('limits').innerHTML = await r.text();
+        document.getElementById('limits').innerHTML = html;
         placeGear();
       }}
     }} catch (e) {{}}
@@ -1106,8 +1107,8 @@ def render_page() -> str:
   }} catch (e) {{}}
   placeGear();
 
-  // раз в минуту перезагружаем страницу (если сообщение не открыто) — подхватить новую версию службы
+  // раз в минуту перезагружаем страницу (если не открыты сообщение и выбор фона) — подхватить новую версию службы
-  setInterval(() => {{ if (!cur) location.reload(); }}, 60000);
+  setInterval(() => {{ if (!cur && !bgrow.classList.contains('open')) location.reload(); }}, 60000);
 </script>
 </body></html>"""
 
@@ -1436,6 +1437,20 @@ async def handle(reader, writer):
         request = (await reader.read(2048)).decode("latin-1")
         target = request.split(" ", 2)[1] if request.count(" ") >= 2 else "/"
         url = urllib.parse.urlparse(target)
+        headers = {}
+        for line in request.split("\r\n")[1:]:
+            name, sep, value = line.partition(":")
+            if sep:
+                headers[name.strip().lower()] = value.strip()
+        port = writer.get_extra_info("sockname")[1]
+        # защита от DNS rebinding и чужих страниц в браузере: только свой адрес в Host,
+        # а действия (/activate, /open-wt) — ещё и не из чужого сайта
+        if (headers.get("host", "").lower() not in (f"127.0.0.1:{port}", f"localhost:{port}")
+                or (url.path in ("/activate", "/open-wt")
+                    and headers.get("sec-fetch-site", "").lower() == "cross-site")):
+            writer.write(b"HTTP/1.1 403 Forbidden\r\nConnection: close\r\nContent-Length: 0\r\n\r\n")
+            await writer.drain()
+            return
         ctype = "text/html; charset=utf-8"
         if url.path == "/limits":
             body = render_limits()
@@ -1520,6 +1535,10 @@ async def register_toolbelt(connection):
     load_config()
     if not port_alive(PORT):
         kickstart_service()
+        for _ in range(20):  # до ~5 с ждём, пока служба займёт порт, иначе web view покажет ошибку
+            if port_alive(PORT):
+                break
+            await asyncio.sleep(0.25)
     await iterm2.tool.async_register_web_view_tool(
         connection,
         display_name="Claude",

tests/test_appearance.py

@@ -29,6 +29,19 @@ class Page(unittest.TestCase):
         self.assertGreater(html.index('id="bgctl"'), limits_end)
 
 
+    def test_reload_skipped_while_colour_row_open(self):
+        html = ca.render_page()
+        self.assertIn("if (!cur && !bgrow.classList.contains('open')) location.reload()", html)
+        self.assertLess(html.index("const gear = document.getElementById('gear'), bgrow"),
+                        html.index("location.reload()"))
+
+    def test_limits_refresh_reads_body_before_moving_gear(self):
+        html = ca.render_page()
+        start = html.index("async function refreshLimits")
+        body = html[start:html.index("setInterval(refreshLimits")]
+        self.assertLess(body.index("await r.text()"), body.index("appendChild(gear)"))
+
+
 class Limits(unittest.TestCase):
     def setUp(self):
         self.saved = dict(ca.state)

tests/test_service.py

@@ -3,8 +3,10 @@ import json
 import os
 import socket
 import sys
+import tempfile
 import types
 import unittest
+import urllib.error
 import urllib.request
 from unittest import mock
 
@@ -40,6 +42,7 @@ class Serve(unittest.TestCase):
         patches = [
             mock.patch.object(ca, "PORT", self.port),
             mock.patch.object(ca, "CONFIG_PATH", "/nonexistent/claude-agents/config.json"),
+            mock.patch.object(ca, "LOG_DIR", self._tmpdir()),
             mock.patch.object(ca, "load_sessions", lambda: []),
             mock.patch.object(ca, "fetch_usage", lambda: {}),
             mock.patch.object(ca, "fetch_balance", lambda: ({}, "test")),
@@ -48,6 +51,11 @@ class Serve(unittest.TestCase):
             p.start()
             self.addCleanup(p.stop)
 
+    def _tmpdir(self):
+        d = tempfile.TemporaryDirectory()
+        self.addCleanup(d.cleanup)
+        return d.name
+
     async def _with_server(self, check):
         task = asyncio.create_task(ca.serve())
         loop = asyncio.get_running_loop()
@@ -91,6 +99,43 @@ class Serve(unittest.TestCase):
         with mock.patch.object(ca, "fetch_usage", no_token):
             asyncio.run(self._with_server(check))
 
+    def test_host_header_checked(self):
+        def get(host, path="/", **extra):
+            req = urllib.request.Request(f"http://127.0.0.1:{self.port}{path}", headers={"Host": host, **extra})
+            try:
+                with urllib.request.urlopen(req, timeout=3) as r:
+                    return r.status
+            except urllib.error.HTTPError as e:
+                e.close()
+                return e.code
+
+        async def check(loop):
+            self.assertEqual(await loop.run_in_executor(None, get, "evil.example:%d" % self.port), 403)
+            self.assertEqual(await loop.run_in_executor(None, get, "127.0.0.1"), 403)
+            self.assertEqual(await loop.run_in_executor(None, get, "127.0.0.1:%d" % self.port), 200)
+            self.assertEqual(await loop.run_in_executor(None, get, "LocalHost:%d" % self.port), 200)
+            self.assertEqual(await loop.run_in_executor(None, get, "evil.example:%d" % self.port, "/raw"), 403)
+        asyncio.run(self._with_server(check))
+
+    def test_cross_site_actions_forbidden(self):
+        def get(path, site):
+            req = urllib.request.Request(f"http://127.0.0.1:{self.port}{path}", headers={"Sec-Fetch-Site": site})
+            try:
+                with urllib.request.urlopen(req, timeout=3) as r:
+                    return r.status
+            except urllib.error.HTTPError as e:
+                e.close()
+                return e.code
+
+        async def check(loop):
+            for path in ("/activate?id=x", "/open-wt?id=x&path=/tmp"):
+                self.assertEqual(await loop.run_in_executor(None, get, path, "cross-site"), 403)
+        with mock.patch.object(ca, "activate", mock.AsyncMock()) as act, \
+                mock.patch.object(ca, "open_worktree", mock.AsyncMock()) as wt:
+            asyncio.run(self._with_server(check))
+        act.assert_not_called()
+        wt.assert_not_called()
+
     def test_serve_busy_port_raises(self):
         s = socket.socket()
         s.bind(("127.0.0.1", self.port))
@@ -125,11 +170,26 @@ class RegisterToolbelt(unittest.TestCase):
             "reveal_if_already_registered": False, "url": "http://127.0.0.1:8766/"}])
 
     def test_register_toolbelt_kickstarts_when_port_dead(self):
-        with mock.patch.object(ca, "port_alive", lambda *a, **k: False):
+        async def no_sleep(sec):
+            pass
+        with mock.patch.object(ca, "port_alive", lambda *a, **k: False), \
+                mock.patch.object(ca.asyncio, "sleep", no_sleep):
             asyncio.run(ca.register_toolbelt(object()))
         self.assertEqual(self.kicks, [1])
         self.assert_registered()
 
+    def test_register_toolbelt_waits_for_service(self):
+        alive = iter([False, False, False, True])  # проверка в начале + ожидание после kickstart
+        sleeps = []
+
+        async def fake_sleep(sec):
+            sleeps.append(sec)
+        with mock.patch.object(ca, "port_alive", lambda *a, **k: next(alive)), \
+                mock.patch.object(ca.asyncio, "sleep", fake_sleep):
+            asyncio.run(ca.register_toolbelt(object()))
+        self.assertEqual(sleeps, [0.25, 0.25])
+        self.assert_registered()
+
     def test_register_toolbelt_skips_kickstart_when_alive(self):
         with mock.patch.object(ca, "port_alive", lambda *a, **k: True):
             asyncio.run(ca.register_toolbelt(object()))