n8n 連携: 最終レビューの指摘を修正

- 進行中の判定: ローカルのコマンドを待ちにしない、ツールの結果待ちを進行中にする、 24 時間更新の無いログは進行中にしない - 写しの区間: 同着は古いファイルを残し、写し元の source でも送信済みにする。送信後に区間が変わったら知らせる - 他のサイトからの API 要求を断る (Host / Origin / Content-Type)。 URL のホストを変えたら合言葉を消す。合言葉は半角の表示文字に限る - n8n との通信の想定外の例外 (送れない文字など) を N8nError にする Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-13 06:36:20 +09:00
parent 929f702cc6
commit 95943f4e0e
5 changed files with 249 additions and 19 deletions
@@ -407,5 +407,172 @@ class RedmineSearchTest(unittest.TestCase):
        self.assertEqual(self.client.search("404"), [])


+def write_raw(root, project_dir, sid, lines, mtime):
+    """jsonl の行 (dict) をそのまま書く。mtime を指定する。"""
+    import os
+    folder = Path(root) / project_dir
+    folder.mkdir(parents=True, exist_ok=True)
+    path = folder / f"{sid}.jsonl"
+    with path.open("w", encoding="utf-8") as fh:
+        for obj in lines:
+            # 本物のログと同じく区切りの空白なしで書く (tool_result 行の高速判定がこの形を前提にしている)
+            fh.write(json.dumps(obj, ensure_ascii=False, separators=(",", ":")) + "\n")
+    os.utime(path, (mtime, mtime))
+    return path
+
+
+def ts(minute):
+    return chktime.iso(BASE + minute * MIN)
+
+
+def user_line(minute, text, sid="p1", sidechain=False):
+    return {"type": "user", "timestamp": ts(minute), "sessionId": sid, "cwd": r"C:\debug\demo",
+            "isSidechain": sidechain, "message": {"role": "user", "content": text}}
+
+
+def reply_line(minute, sid="p1", tool=None):
+    content = [{"type": "tool_use", "id": "t1", "name": tool, "input": {}}] if tool else [{"type": "text", "text": "ok"}]
+    return {"type": "assistant", "timestamp": ts(minute), "sessionId": sid, "message": {"model": "m", "content": content}}
+
+
+def tool_result_line(minute, sid="p1"):
+    return {"type": "user", "timestamp": ts(minute), "sessionId": sid,
+            "message": {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "t1", "content": "done"}]}}
+
+
+class FinalReviewFixTest(unittest.TestCase):
+    """最終レビューで見つかった問題の再現テスト。"""
+
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.root = Path(self.tmp.name) / "projects"
+        self.store = chktime.Store(self.root)
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def blocks(self, now):
+        return chktime.collect_blocks(self.store, "", date(2026, 1, 5), date(2026, 1, 5), 30, 5, now=now)
+
+    # Important 1: 最後がローカルのコマンド (/exit など) だと永遠に進行中になっていた
+    def test_trailing_local_command_is_not_in_progress(self):
+        write_raw(self.root, "C--debug-demo", "p1",
+                  [user_line(0, "作業して"), reply_line(20), user_line(21, "<command-name>/exit</command-name>")],
+                  BASE + 21 * MIN)
+        self.assertEqual([b["inProgress"] for b in self.blocks(LATER)], [False])
+
+    def test_prompt_without_reply_long_ago_is_not_in_progress(self):
+        write_raw(self.root, "C--debug-demo", "p1", [user_line(0, "作業して"), reply_line(20), user_line(21, "次")],
+                  BASE + 21 * MIN)
+        self.assertEqual([b["inProgress"] for b in self.blocks(BASE + 21 * MIN + 25 * 3600)], [False])
+        self.assertEqual([b["inProgress"] for b in self.blocks(BASE + 21 * MIN + 2 * 3600)], [True])
+
+    # Important 2: ツールの結果を待っている間に 30 分静かだと、途中で切れた区間が送れていた
+    def test_open_tool_call_keeps_block_in_progress(self):
+        write_raw(self.root, "C--debug-demo", "p1", [user_line(0, "調べて"), reply_line(5, tool="Agent")],
+                  BASE + 5 * MIN)
+        self.assertEqual([b["inProgress"] for b in self.blocks(BASE + 40 * MIN)], [True])
+        write_raw(self.root, "C--debug-demo", "p1",
+                  [user_line(0, "調べて"), reply_line(5, tool="Agent"), tool_result_line(45), reply_line(46)],
+                  BASE + 46 * MIN)
+        self.store.clear()
+        self.assertEqual([(b["inProgress"], b["minutes"]) for b in self.blocks(BASE + 100 * MIN)], [(False, 46)])
+
+    def test_sent_block_that_grew_is_flagged(self):
+        block = {"source": "chktime:a:1", "startEpoch": BASE, "endEpoch": BASE + 46 * MIN, "start": ts(0), "end": ts(46),
+                 "taskName": "A"}
+        rows = [{"id": 1, "source": "chktime:a:1", "start_time": ts(0), "end_time": ts(5), "task": "A"}]
+        chktime.mark_blocks([block], rows, [], now=LATER)
+        self.assertTrue(block["sent"])
+        self.assertTrue(block["changedSinceSent"])
+
+    # Important 3: 写しの同着で、送信済みの区間が未送信に見えていた
+    def test_copy_with_same_end_keeps_sent_state(self):
+        lines_old = [user_line(0, "設計して", sid="aaa"), reply_line(10, sid="aaa")]
+        lines_new = [user_line(0, "設計して", sid="zzz"), reply_line(10, sid="zzz"), user_line(60, "続き", sid="zzz"),
+                     reply_line(70, sid="zzz")]
+        write_raw(self.root, "C--debug-demo", "zzz", lines_new, BASE + 70 * MIN)
+        write_raw(self.root, "C--debug-demo", "aaa", lines_old, BASE + 10 * MIN)
+        blocks = self.blocks(LATER)
+        first = blocks[0]
+        self.assertEqual(len(blocks), 2)
+        sent_source = f"chktime:aaa:{int(round(BASE * 1000))}"
+        self.assertIn(sent_source, [first["source"]] + first["aliases"])
+        rows = [{"id": 1, "source": sent_source, "start_time": ts(0), "end_time": ts(10), "task": "A"}]
+        chktime.mark_blocks(blocks, rows, [], now=LATER)
+        self.assertTrue(first["sent"])
+        self.assertEqual(first["overlaps"], [])
+
+    # Important 5: 送れない文字の合言葉で一覧ごと 500 になっていた
+    def test_non_ascii_token_is_n8n_error_without_token_text(self):
+        client = chktime.N8n("http://127.0.0.1:9", "合言葉です")
+        with self.assertRaises(chktime.N8nError) as ctx:
+            client.rows(date(2026, 1, 1), date(2026, 1, 1))
+        self.assertNotIn("合言葉です", str(ctx.exception))
+
+    def test_holder_rejects_non_ascii_token(self):
+        holder = chktime.N8nHolder(Path(self.tmp.name) / "n8n.json")
+        with self.assertRaises(chktime.N8nError):
+            holder.update({"token": "合言葉"})
+
+    # Important 4: URL だけ変えると、保存済みの合言葉が別のサーバーへ送られていた
+    def test_changing_url_host_clears_token(self):
+        holder = chktime.N8nHolder(Path(self.tmp.name) / "n8n.json")
+        holder.update({"url": "https://n8n.example", "token": "secret"})
+        self.assertFalse(holder.update({"url": "https://attacker.example"})["hasToken"])
+        holder.update({"url": "https://n8n.example", "token": "secret"})
+        self.assertTrue(holder.update({"url": "https://n8n.example/"})["hasToken"])
+        self.assertTrue(holder.update({"url": "https://other.example", "token": "secret2"})["hasToken"])
+
+
+class PostGuardTest(unittest.TestCase):
+    """Important 4: 他のサイトから POST で設定を書き換えられないこと (Content-Type と Origin)。"""
+
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        server = chktime.Server(("127.0.0.1", 0), chktime.Handler)
+        server.store = chktime.Store(Path(self.tmp.name))
+        server.redmine = chktime.RedmineHolder(Path(self.tmp.name) / "redmine.json")
+        server.n8n = chktime.N8nHolder(Path(self.tmp.name) / "n8n.json")
+        server.jobs = chktime.Jobs()
+        self.server = server
+        self.port = server.server_address[1]
+        threading.Thread(target=server.serve_forever, daemon=True).start()
+
+    def tearDown(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.tmp.cleanup()
+
+    def post(self, body, ctype="application/json", origin=None, host=None):
+        import http.client
+        conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
+        headers = {"Content-Type": ctype}
+        if origin:
+            headers["Origin"] = origin
+        if host:
+            headers["Host"] = host
+        conn.request("POST", "/api/n8n/config", body=json.dumps(body).encode(), headers=headers)
+        res = conn.getresponse()
+        status = res.status
+        res.read()
+        conn.close()
+        return status
+
+    def test_rejects_simple_cross_site_post(self):
+        self.assertEqual(self.post({"url": "https://attacker.example"}, ctype="text/plain"), 415)
+        self.assertEqual(self.server.n8n.public()["url"], "")
+
+    def test_rejects_foreign_origin_and_host(self):
+        self.assertEqual(self.post({"url": "https://a.example"}, origin="https://evil.example"), 403)
+        self.assertEqual(self.post({"url": "https://a.example"}, host="evil.example"), 403)
+        self.assertEqual(self.server.n8n.public()["url"], "")
+
+    def test_accepts_same_origin_json(self):
+        origin = f"http://127.0.0.1:{self.port}"
+        self.assertEqual(self.post({"url": "https://n8n.example"}, origin=origin), 200)
+        self.assertEqual(self.server.n8n.public()["url"], "https://n8n.example")
+
+
 if __name__ == "__main__":
    unittest.main()