# Files changed · PR \#368 · lucenx/codexbar-plasma [View on GitCafe](https://git.cafe/lucenx/codexbar-plasma/pull/368/diffs) Repository: [lucenx/codexbar-plasma](https://git.cafe/lucenx/codexbar-plasma) Visibility: public [fix\(todo-gate\): refuse redirects, bound reads and git probes](https://git.cafe/lucenx/codexbar-plasma/pull/368.md) Version: 1 Head commit: 90c873030bd4694e86a8252037e55b6333bbff25 Base commit: 6e93105bdab859b335b55ff4fe3cffdc28a45bf8 Comparison base: 6e93105bdab859b335b55ff4fe3cffdc28a45bf8 Showing up to 20 entries from this revision. If the pull request changes, reopen the first page before continuing. ## CHANGELOG.md Status: modified; +3; −0 ```diff @@ -12,6 +12,9 @@ ### Fixed +- Harden the TODO gate helper: it refuses Decisions redirects instead of + following them with the API key, rejects oversized responses, and bounds + its git probes with a timeout. - Keep the stored AI Insights model when opening settings. Opening and saving the page no longer clears the model, so AI Insights stops asking for reconfiguration when nothing was changed. The stored model is also offered ``` ## scripts/todo_gate.py Status: modified; +19; −4 ```diff @@ -14,6 +14,9 @@ ENDPOINT = "https://openrouter.ai/api/alpha/decisions" MODEL = "~typesafe/jev-latest" API_KEY_ENV = "OPENROUTER_API_KEY" +# The verdict is one float plus usage; anything larger is not an answer. +MAX_RESPONSE_BYTES = 65536 +GIT_TIMEOUT = 120 # Jev accepts 32k tokens. Cap both inputs well below that and mark truncation, # so a dropped hunk or entry is never read as absent. DIFF_LIMIT = 60000 @@ -89,7 +92,8 @@ """Return the committed paths, diff, and TODO.md for this branch.""" def git(*args): return subprocess.check_output(["git", *args], cwd=root, text=True, + stderr=subprocess.DEVNULL, + timeout=GIT_TIMEOUT) - stderr=subprocess.DEVNULL) paths = [path for path in git("diff", "--merge-base", base, "HEAD", "--name-only").splitlines() if path] @@ -102,14 +106,25 @@ return paths, git("diff", "--merge-base", base, "HEAD"), todo +class NoRedirect(urllib.request.HTTPRedirectHandler): + """A redirect could carry the Authorization header to another host.""" + + def redirect_request(self, req, fp, code, msg, headers, newurl): + return None + + def ask(request, key): """Post the Decisions request and return its answers and usage.""" call = urllib.request.Request( ENDPOINT, data=json.dumps(request).encode(), headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"}, ) + opener = urllib.request.build_opener(NoRedirect) + with opener.open(call, timeout=120) as response: + raw = response.read(MAX_RESPONSE_BYTES + 1) + if len(raw) > MAX_RESPONSE_BYTES: + raise ValueError("the Decisions response exceeded the supported size") - with urllib.request.urlopen(call, timeout=120) as response: + payload = json.loads(raw) - payload = json.load(response) if not isinstance(payload, dict) or not isinstance(payload.get("answers"), dict): raise ValueError("the Decisions response carried no answers") usage = payload.get("usage") @@ -141,7 +156,7 @@ return 0 try: paths, diff, todo = read_change(options.base) - except (OSError, subprocess.CalledProcessError): + except (OSError, subprocess.SubprocessError): print(f"cannot diff against {options.base}; skipping the TODO gate") return 0 if not paths: ``` ## tests/test_todo_gate.py Status: modified; +105; −2 ```diff @@ -1,12 +1,20 @@ """The TODO gate must send TODO.md, flag truncation, and reject malformed answers.""" +import http.server +import json from pathlib import Path +import subprocess import sys +import threading import unittest +import urllib.error +from unittest.mock import patch sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts")) -from todo_gate import (DIFF_LIMIT, TODO_LIMIT, build_request, probability, +import todo_gate +from todo_gate import (DIFF_LIMIT, TODO_LIMIT, ask, build_request, main, + probability, read_change, threshold_argument, + usage_line, verdict) - threshold_argument, usage_line, verdict) def answered(probability): @@ -89,5 +97,100 @@ self.assertIn("input tokens", usage_line(usage)) +class GateHandler(http.server.BaseHTTPRequestHandler): + mode = "ok" + evil_hits = 0 + evil_authorization = None + + def log_message(self, *args): + pass + + def _send_json(self, code, payload, location=None): + raw = json.dumps(payload).encode() + self.send_response(code) + if location is not None: + self.send_header("Location", location) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(raw))) + self.end_headers() + self.wfile.write(raw) + + def do_POST(self): + length = int(self.headers.get("Content-Length") or 0) + if length: + self.rfile.read(length) + if GateHandler.mode == "redirect": + self._send_json(302, {}, location="/evil") + elif GateHandler.mode == "big": + self._send_json(200, {"answers": {"todo": {"noul": 0.1}}, + "pad": "x" * 70000}) + else: + self._send_json(200, {"answers": {"todo": {"noul": 0.1}}, + "usage": {}}) + + def do_GET(self): + if self.path == "/evil": + GateHandler.evil_hits += 1 + GateHandler.evil_authorization = self.headers.get("Authorization") + self._send_json(200, {"answers": {"todo": {"noul": 0.99}}, + "usage": {}}) + + +class AskTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), + GateHandler) + cls.thread = threading.Thread(target=cls.server.serve_forever, + daemon=True) + cls.thread.start() + cls.url = f"http://127.0.0.1:{cls.server.server_address[1]}" + + @classmethod + def tearDownClass(cls): + cls.server.shutdown() + cls.server.server_close() + + def setUp(self): + GateHandler.mode = "ok" + GateHandler.evil_hits = 0 + GateHandler.evil_authorization = None + endpoint = patch.object(todo_gate, "ENDPOINT", self.url) + endpoint.start() + self.addCleanup(endpoint.stop) + + def test_small_valid_response_is_accepted(self): + answers, usage = ask({"model": "m"}, "key") + self.assertEqual(answers, {"todo": {"noul": 0.1}}) + self.assertEqual(usage, {}) + + def test_redirect_is_refused_without_sending_the_key(self): + GateHandler.mode = "redirect" + with self.assertRaises(urllib.error.HTTPError): + ask({"model": "m"}, "key") + self.assertEqual(GateHandler.evil_hits, 0) + self.assertIsNone(GateHandler.evil_authorization) + + def test_oversized_response_is_rejected(self): + GateHandler.mode = "big" + with self.assertRaises(ValueError): + ask({"model": "m"}, "key") + + +class ReadChangeTests(unittest.TestCase): + def test_git_calls_carry_a_timeout(self): + with patch("subprocess.check_output", return_value="") as git: + read_change("main") + self.assertTrue(git.called) + for call in git.call_args_list: + self.assertEqual(call.kwargs.get("timeout"), 120) + + def test_git_timeout_skips_the_gate(self): + with patch.object(todo_gate, "read_change", + side_effect=subprocess.TimeoutExpired("git", 120)), \ + patch.dict("os.environ", {"OPENROUTER_API_KEY": "key"}): + self.assertEqual(main(["--base", "main"]), 0) + + if __name__ == "__main__": unittest.main() ```