fix: guard CGI Status-line parse in _write_cgi_response

An empty or non-numeric Status: header from git http-backend raised
ValueError/IndexError that escaped the handler thread. Wrap the parse
in a try/except and fall back to HTTP 500 instead.

Closes #254

.coveragerc

@@ -1,9 +0,0 @@
-[run]
-branch = True
-source = .
-
-[report]
-omit =
-    bot_bottle/egress_addon.py
-    bot_bottle/cli/tui.py
-    tests/*

.gitea/workflows/test.yml

@@ -39,14 +39,8 @@ jobs:
         with:
           python-version: "3.12"
 
-      - name: Install dev requirements
-        run: python3 -m pip install -r requirements-dev.txt
-
       - name: Run unit tests
-        run: python3 -m coverage run -m unittest discover -t . -s tests/unit -v
-
-      - name: Report unit coverage
-        run: python3 -m coverage report -m
+        run: python3 -m unittest discover -t . -s tests/unit -v
 
   integration:
     runs-on: ubuntu-latest

.gitignore

@@ -22,4 +22,3 @@ venv/
 .pytest_cache/
 .mypy_cache/
 .ruff_cache/
-.coverage

bot_bottle/cli/supervise.py

@@ -2,8 +2,9 @@
 act on them (approve / modify / reject).
 
 Curses-based TUI; modify-then-approve shells out to $EDITOR. The
-Egress proposals are queued for operator review as full routes.yaml
-updates.
+approval handler wires to PRD 0016 (capability-block), which rebuilds
+the bottle Dockerfile. Egress proposals are queued for operator review
+as full routes.yaml updates.
 """
 
 from __future__ import annotations
@@ -21,6 +22,10 @@ from pathlib import Path
 
 from .. import supervise as _supervise
 from ..bottle_state import read_metadata
+# from ..backend.docker.capability_apply import (
+#     CapabilityApplyError,
+#     apply_capability_change,
+# )
 from ..backend.docker.egress_apply import (
     EgressApplyError,
     applicator as _docker_applicator,
@@ -33,6 +38,10 @@ from ..backend.smolmachines.egress_apply import (
 )
 from ..log import Die, error, info
 
+
+class CapabilityApplyError(RuntimeError):
+    """Placeholder while capability_apply is disabled."""
+
 from ..supervise import (
     COMPONENT_FOR_TOOL,
     AuditEntry,
@@ -41,10 +50,12 @@ from ..supervise import (
     STATUS_APPROVED,
     STATUS_MODIFIED,
     STATUS_REJECTED,
+    TOOL_CAPABILITY_BLOCK,
     TOOL_EGRESS_ALLOW,
     TOOL_EGRESS_BLOCK,
     TOOL_GITLEAKS_ALLOW,
     TOOL_EGRESS_TOKEN_ALLOW,
+    archive_proposal,
     list_pending_proposals,
     render_diff,
     write_audit_entry,
@@ -72,7 +83,7 @@ class QueuedProposal:
 # Errors any remediation engine may raise. Caught by the TUI key
 # handlers and surfaced in the status line so a failed apply keeps
 # the proposal pending rather than crashing curses.
-ApplyError = (EgressApplyError,)
+ApplyError = (CapabilityApplyError, EgressApplyError)
 
 
 def apply_routes_change(slug: str, content: str) -> tuple[str, str]:
@@ -132,6 +143,8 @@ def _detail_lines(
 
 
 def _suffix_for_tool(tool: str) -> str:
+    if tool == TOOL_CAPABILITY_BLOCK:
+        return ".dockerfile"
     if tool in (TOOL_EGRESS_ALLOW, TOOL_EGRESS_BLOCK):
         return ".yaml"
     if tool in (TOOL_GITLEAKS_ALLOW, TOOL_EGRESS_TOKEN_ALLOW):
@@ -153,6 +166,17 @@ def approve(
     file_to_apply = final_file if final_file is not None else qp.proposal.proposed_file
 
     diff_before, diff_after = "", ""
+    # if qp.proposal.tool == TOOL_CAPABILITY_BLOCK:
+    #     _meta = read_metadata(qp.proposal.bottle_slug)
+    #     if _meta is not None and not _meta.compose_project:
+    #         raise CapabilityApplyError(
+    #             "capability-block remediation is not supported for smolmachines "
+    #             "bottles. Reject this proposal or handle the capability change "
+    #             "manually, then restart the bottle."
+    #         )
+    #     diff_before, diff_after = apply_capability_change(
+    #         qp.proposal.bottle_slug, file_to_apply,
+    #     )
     if qp.proposal.tool in (TOOL_EGRESS_ALLOW, TOOL_EGRESS_BLOCK):
         diff_before, diff_after = apply_routes_change(
             qp.proposal.bottle_slug,
@@ -170,6 +194,9 @@ def approve(
         qp, action=status, notes=notes,
         diff_before=diff_before, diff_after=diff_after,
     )
+    if qp.proposal.tool == TOOL_CAPABILITY_BLOCK:
+        archive_proposal(qp.queue_dir, qp.proposal.id)
+
 
 def reject(qp: QueuedProposal, *, reason: str) -> None:
     """Write a rejection response and an audit entry."""
@@ -319,7 +346,7 @@ def _list_once() -> int:
     return 0
 
 
-def _try_init_green() -> int:  # pragma: no cover
+def _try_init_green() -> int:
     """Initialise a green color pair and return its attr, or 0."""
     try:
         curses.start_color()
@@ -330,7 +357,7 @@ def _try_init_green() -> int:  # pragma: no cover
         return 0
 
 
-def _main_loop(stdscr: "curses._CursesWindow") -> None:  # type: ignore  # pragma: no cover
+def _main_loop(stdscr: "curses._CursesWindow") -> None:  # type: ignore
     curses.curs_set(0)
     stdscr.timeout(_REFRESH_INTERVAL_MS)
     green_attr = _try_init_green()
@@ -420,7 +447,7 @@ def _render(
     status_line: str,
     *,
     green_attr: int = 0,  # noqa: F841 — unused, but required by interface
-) -> None:  # pragma: no cover
+) -> None:
     stdscr.erase()
     h, w = stdscr.getmaxyx()
     header = f"bot-bottle supervise  ({len(pending)} pending)"
@@ -471,7 +498,7 @@ def _detail_view(
     qp: QueuedProposal,
     *,
     green_attr: int = 0,
-) -> None:  # pragma: no cover
+) -> None:
     """Render the full proposal. Scrollable. Press q to return."""
     lines = _detail_lines(qp, green_attr=green_attr)
     offset = 0
@@ -523,7 +550,7 @@ def _detail_view(
             return
 
 
-def _modify(stdscr: "curses._CursesWindow", qp: QueuedProposal) -> str | None:  # type: ignore  # pragma: no cover
+def _modify(stdscr: "curses._CursesWindow", qp: QueuedProposal) -> str | None:  # type: ignore
     """Suspend curses, open $EDITOR on the proposed file, return edited content."""
     suffix = _suffix_for_tool(qp.proposal.tool)
     curses.endwin()
@@ -534,7 +561,7 @@ def _modify(stdscr: "curses._CursesWindow", qp: QueuedProposal) -> str | None:
     return edited
 
 
-def _prompt(stdscr: "curses._CursesWindow", label: str) -> str:  # type: ignore  # pragma: no cover
+def _prompt(stdscr: "curses._CursesWindow", label: str) -> str:  # type: ignore
     """One-line input at the bottom of the screen."""
     curses.curs_set(1)
     h, _ = stdscr.getmaxyx()

bot_bottle/git_http_backend.py

@@ -148,7 +148,13 @@ class GitHttpHandler(BaseHTTPRequestHandler):
             key, _, value = line.decode("latin1").partition(":")
             value = value.strip()
             if key.lower() == "status":
-                status = int(value.split()[0])
+                try:
+                    status = int(value.split()[0])
+                except (ValueError, IndexError):
+                    self.log_message(
+                        "malformed CGI Status header %r; using 500", value,
+                    )
+                    status = 500
             else:
                 headers.append((key, value))
         self.send_response(status)

requirements-dev.txt

@@ -4,4 +4,3 @@
 
 pylint>=3.0.0
 pyright>=1.1.300
-coverage>=7.0.0

tests/integration/test_sandbox_escape.py

@@ -92,9 +92,9 @@ class TestSandboxEscape(unittest.TestCase):
                     "on PATH: curl -sSL https://smolmachines.com/install.sh | sh"
                 )
 
-        # Throwaway static key for the git-gate fixture. It need not
-        # be a real SSH key: test 5 reaches gitleaks before any SSH
-        # attempt anyway.
+        # Throwaway "identity file" for the git-gate's `identity` field.
+        # It need not be a real SSH key: test 5 reaches gitleaks before
+        # any SSH attempt anyway.
         fd, kp = tempfile.mkstemp(prefix="sandbox-test-key.")
         os.close(fd)
         cls._key_path = Path(kp)
@@ -123,10 +123,7 @@ class TestSandboxEscape(unittest.TestCase):
                     "git-gate": {"repos": {
                         "throwaway": {
                             "url": "ssh://git@unreachable.invalid:22/throwaway.git",
-                            "key": {
-                                "provider": "static",
-                                "path": str(cls._key_path),
-                            },
+                            "identity": str(cls._key_path),
                         },
                     }},
                 },

tests/integration/test_smolmachines_launch.py

@@ -198,7 +198,6 @@ class TestSmolmachinesLaunch(unittest.TestCase):
         # connect fails, which is the property chunk 3 will
         # preserve once egress is actually running.
         r = self.bottle.exec(
-            "env -u HTTPS_PROXY -u HTTP_PROXY -u https_proxy -u http_proxy "
             f"curl -s --show-error --max-time 3 http://{self.plan.bundle_ip}:9099 "
             "2>&1 || true"
         )

tests/unit/test_git_http_backend.py

@@ -256,6 +256,57 @@ class TestGitHttpBackend(unittest.TestCase):
             os.environ["GIT_GATE_ACCESS_HOOK"] = value
 
 
+class TestMalformedStatusHeader(unittest.TestCase):
+    """Malformed CGI Status: headers must not propagate as unhandled exceptions;
+    the handler should fall back to HTTP 500."""
+
+    def setUp(self):
+        from http.server import ThreadingHTTPServer
+        import tempfile
+        self._tmp = tempfile.mkdtemp()
+        os.environ["GIT_PROJECT_ROOT"] = self._tmp
+        self._server = ThreadingHTTPServer(("127.0.0.1", 0), GitHttpHandler)
+        self._thread = threading.Thread(
+            target=self._server.serve_forever, daemon=True,
+        )
+        self._thread.start()
+        self._port = self._server.server_port
+
+    def tearDown(self):
+        self._server.shutdown()
+        self._server.server_close()
+        os.environ.pop("GIT_PROJECT_ROOT", None)
+        import shutil
+        shutil.rmtree(self._tmp, ignore_errors=True)
+
+    def _get_with_backend_response(self, cgi_response: bytes) -> int:
+        with mock.patch(
+            "bot_bottle.git_http_backend.subprocess.run",
+            return_value=mock.Mock(returncode=0, stdout=cgi_response),
+        ):
+            req = urllib.request.Request(
+                f"http://127.0.0.1:{self._port}/repo.git/info/refs",
+                method="GET",
+            )
+            try:
+                with urllib.request.urlopen(req, timeout=3) as resp:
+                    return resp.status
+            except urllib.error.HTTPError as e:  # type: ignore
+                return e.code
+
+    def test_empty_status_value_returns_500(self):
+        status = self._get_with_backend_response(
+            b"Status: \r\nContent-Type: text/plain\r\n\r\n"
+        )
+        self.assertEqual(500, status)
+
+    def test_non_numeric_status_returns_500(self):
+        status = self._get_with_backend_response(
+            b"Status: bad\r\nContent-Type: text/plain\r\n\r\n"
+        )
+        self.assertEqual(500, status)
+
+
 class TestContentLengthBounds(unittest.TestCase):
     """PRD 0041: malformed or oversized Content-Length is rejected before
     git http-backend is invoked."""

tests/unit/test_supervise_server.py

@@ -20,7 +20,6 @@ import supervise as _sv  # noqa: E402  # type: ignore
 
 from bot_bottle import supervise_server  # noqa: E402
 from bot_bottle.supervise_server import (
-    ERR_INTERNAL,
     ERR_INVALID_PARAMS,
     ERR_INVALID_REQUEST,
     ERR_METHOD_NOT_FOUND,
@@ -30,9 +29,7 @@ from bot_bottle.supervise_server import (
     PROPOSED_FILE_FIELD,
     ServerConfig,
     TOOL_DEFINITIONS,
-    _RpcClientError,
     _RpcError,
-    _RpcInternalError,
     _response_timeout_from_env,
     format_response_text,
     handle_initialize,
@@ -50,15 +47,15 @@ from bot_bottle.supervise_server import (
 
 
 class TestValidation(unittest.TestCase):
+    def test_capability_block_accepts_anything_nonempty(self):
+        validate_proposed_file(
+            _sv.TOOL_CAPABILITY_BLOCK,
+            "FROM python:3.13\nRUN apk add git\n",
+        )
+
     def test_empty_proposed_file_rejected_for_tools_with_file_field(self):
         with self.assertRaises(_RpcError):
-            validate_proposed_file(_sv.TOOL_EGRESS_ALLOW, "   \n\t")
-
-    def test_capability_block_rejected_as_unknown_tool(self):
-        with self.assertRaises(_RpcError) as cm:
-            validate_proposed_file("capability-block", "FROM python:3.13\n")
-        self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code)
-        self.assertIn("unknown tool", cm.exception.message)
+            validate_proposed_file(_sv.TOOL_CAPABILITY_BLOCK, "   \n\t")
 
     def test_egress_routes_yaml_is_validated(self):
         validate_proposed_file(
@@ -80,65 +77,6 @@ class TestValidation(unittest.TestCase):
         self.assertIn("must not change egress logging", cm.exception.message)
 
 
-# --- Error taxonomy --------------------------------------------------------
-
-
-class TestRpcErrorTaxonomy(unittest.TestCase):
-    def test_rpc_client_error_is_rpc_error(self):
-        e = _RpcClientError(ERR_INVALID_PARAMS, "bad param")
-        self.assertIsInstance(e, _RpcError)
-        self.assertEqual(ERR_INVALID_PARAMS, e.code)
-        self.assertEqual("bad param", e.message)
-
-    def test_rpc_internal_error_is_rpc_error(self):
-        e = _RpcInternalError("disk full")
-        self.assertIsInstance(e, _RpcError)
-        self.assertEqual(ERR_INTERNAL, e.code)
-        self.assertEqual("disk full", e.message)
-
-    def test_rpc_internal_error_preserves_cause(self):
-        cause = OSError("no space left on device")
-        try:
-            raise _RpcInternalError("failed to write") from cause
-        except _RpcInternalError as e:
-            self.assertIs(cause, e.__cause__)
-
-    def test_parse_error_is_client_error(self):
-        with self.assertRaises(_RpcClientError):
-            parse_jsonrpc(b"{bad json")
-
-    def test_validation_error_is_client_error(self):
-        with self.assertRaises(_RpcClientError):
-            validate_proposed_file(_sv.TOOL_EGRESS_ALLOW, "routes: nope\n")
-
-    def test_unknown_tool_in_tools_call_is_client_error(self):
-        config = ServerConfig(bottle_slug="dev", queue_dir=Path("/unused"))
-        with self.assertRaises(_RpcClientError) as cm:
-            handle_tools_call({"name": "no-such-tool", "arguments": {}}, config)
-        self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code)
-
-
-class TestRpcInternalErrorOnIoFailure(unittest.TestCase):
-    def test_write_proposal_os_error_raises_internal(self):
-        config = ServerConfig(
-            bottle_slug="dev",
-            queue_dir=Path("/dev/null/cannot-exist"),
-        )
-        with self.assertRaises(_RpcInternalError) as cm:
-            handle_tools_call(
-                {
-                    "name": _sv.TOOL_EGRESS_ALLOW,
-                    "arguments": {
-                        "routes_yaml": "routes:\n  - host: example.com\n",
-                        "justification": "x",
-                    },
-                },
-                config,
-            )
-        self.assertEqual(ERR_INTERNAL, cm.exception.code)
-        self.assertIsNotNone(cm.exception.__cause__)
-
-
 # --- JSON-RPC parsing ------------------------------------------------------
 
 
@@ -219,6 +157,7 @@ class TestHandleToolsList(unittest.TestCase):
         self.assertEqual(
             sorted([
                 _sv.TOOL_EGRESS_ALLOW,
+                _sv.TOOL_CAPABILITY_BLOCK,
                 _sv.TOOL_EGRESS_BLOCK,
                 _sv.TOOL_LIST_EGRESS_ROUTES,
             ]),
@@ -294,10 +233,10 @@ class TestHandleToolsCall(unittest.TestCase):
         try:
             result = handle_tools_call(
                 {
-                    "name": _sv.TOOL_EGRESS_BLOCK,
+                    "name": _sv.TOOL_CAPABILITY_BLOCK,
                     "arguments": {
-                        "routes_yaml": "routes:\n  - host: example.com\n",
-                        "justification": "need example.com",
+                        "dockerfile": "FROM python:3.13\n",
+                        "justification": "need git",
                     },
                 },
                 self.config,
@@ -334,9 +273,9 @@ class TestHandleToolsCall(unittest.TestCase):
         try:
             result = handle_tools_call(
                 {
-                    "name": _sv.TOOL_EGRESS_ALLOW,
+                    "name": _sv.TOOL_CAPABILITY_BLOCK,
                     "arguments": {
-                        "routes_yaml": "routes:\n  - host: example.com\n",
+                        "dockerfile": "FROM python:3.13\n",
                         "justification": "needed for tests",
                     },
                 },
@@ -358,52 +297,20 @@ class TestHandleToolsCall(unittest.TestCase):
         with self.assertRaises(_RpcError):
             handle_tools_call(
                 {
-                    "name": _sv.TOOL_EGRESS_ALLOW,
-                    "arguments": {"routes_yaml": "routes:\n  - host: example.com\n"},
+                    "name": _sv.TOOL_CAPABILITY_BLOCK,
+                    "arguments": {"dockerfile": "FROM python:3.13\n"},
                 },
                 self.config,
             )
 
-    def test_missing_name_raises(self):
-        with self.assertRaises(_RpcError) as cm:
-            handle_tools_call({"arguments": {}}, self.config)
-        self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code)
-
-    def test_arguments_must_be_object(self):
-        with self.assertRaises(_RpcError) as cm:
-            handle_tools_call(
-                {
-                    "name": _sv.TOOL_EGRESS_ALLOW,
-                    "arguments": [],
-                },
-                self.config,
-            )
-        self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code)
-        self.assertIn("must be an object", cm.exception.message)
-
-    def test_capability_block_call_raises_unknown_tool(self):
-        with self.assertRaises(_RpcError) as cm:
-            handle_tools_call(
-                {
-                    "name": "capability-block",
-                    "arguments": {
-                        "dockerfile": "FROM python:3.13\n",
-                        "justification": "need git",
-                    },
-                },
-                self.config,
-            )
-        self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code)
-        self.assertIn("unknown tool", cm.exception.message)
-
     def test_archives_proposal_after_response(self):
         responder = self._respond_when_proposal_appears(_sv.STATUS_APPROVED)
         try:
             handle_tools_call(
                 {
-                    "name": _sv.TOOL_EGRESS_ALLOW,
+                    "name": _sv.TOOL_CAPABILITY_BLOCK,
                     "arguments": {
-                        "routes_yaml": "routes:\n  - host: example.com\n",
+                        "dockerfile": "FROM python:3.13\n",
                         "justification": "x",
                     },
                 },
@@ -425,10 +332,10 @@ class TestHandleToolsCall(unittest.TestCase):
         )
         result = handle_tools_call(
             {
-                "name": _sv.TOOL_EGRESS_ALLOW,
+                "name": _sv.TOOL_CAPABILITY_BLOCK,
                 "arguments": {
-                    "routes_yaml": "routes:\n  - host: example.com\n",
-                    "justification": "need egress",
+                    "dockerfile": "FROM python:3.13\n",
+                    "justification": "need a capability",
                 },
             },
             config,
@@ -443,31 +350,6 @@ class TestHandleToolsCall(unittest.TestCase):
 
 
 class TestHandleListEgressRoutes(unittest.TestCase):
-    def test_success_returns_body_text(self):
-        class _Resp:
-            def __enter__(self):
-                return self
-
-            def __exit__(self, exc_type: type[BaseException] | None, exc: BaseException | None, tb: object) -> bool:
-                return False
-
-            def read(self):
-                return b"[{\"host\": \"example.com\"}]"
-
-        class _Opener:
-            def open(self, *args, **kwargs):  # noqa: ANN001, ANN002, ANN003  # type: ignore
-                return _Resp()
-
-        with patch.object(supervise_server.urllib.request, "build_opener", return_value=_Opener()):
-            result = handle_list_egress_routes(
-                {},
-                ServerConfig(bottle_slug="dev", queue_dir=Path("/unused")),
-            )
-
-        self.assertFalse(result["isError"])  # type: ignore[index]
-        text = result["content"][0]["text"]  # type: ignore[index]
-        self.assertIn("example.com", text)
-
     def test_url_error_returns_tool_error(self):
         class _Opener:
             def open(self, *args, **kwargs):  # noqa: ANN001, ANN002, ANN003  # type: ignore
@@ -527,13 +409,6 @@ class TestFormatResponseText(unittest.TestCase):
         self.assertIn("the operator modified", text.lower())
 
 
-class TestFormatPendingResponseText(unittest.TestCase):
-    def test_formats_timeout_message(self):
-        text = supervise_server.format_pending_response_text(12.5)
-        self.assertIn("status: pending", text)
-        self.assertIn("12.5s", text)
-
-
 # --- End-to-end HTTP sanity ------------------------------------------------
 
 
@@ -584,7 +459,7 @@ class TestHttpEndToEnd(unittest.TestCase):
         self.assertEqual("2.0", result["jsonrpc"])
         self.assertEqual(1, result["id"])
         names = [t["name"] for t in result["result"]["tools"]]  # type: ignore[index]
-        self.assertNotIn("capability-block", names)
+        self.assertIn(_sv.TOOL_CAPABILITY_BLOCK, names)
         self.assertIn(_sv.TOOL_EGRESS_ALLOW, names)
         self.assertIn(_sv.TOOL_EGRESS_BLOCK, names)
 
@@ -594,26 +469,6 @@ class TestHttpEndToEnd(unittest.TestCase):
         )
         self.assertEqual(ERR_METHOD_NOT_FOUND, result["error"]["code"])  # type: ignore[index]
 
-    def test_internal_error_returns_err_internal_over_http(self):
-        with patch.object(
-            supervise_server._sv, "write_proposal",
-            side_effect=OSError("disk full"),
-        ):
-            result = self._post_jsonrpc({
-                "jsonrpc": "2.0",
-                "id": 99,
-                "method": "tools/call",
-                "params": {
-                    "name": _sv.TOOL_EGRESS_ALLOW,
-                    "arguments": {
-                        "routes_yaml": "routes:\n  - host: example.com\n",
-                        "justification": "x",
-                    },
-                },
-            })
-        self.assertIn("error", result)
-        self.assertEqual(ERR_INTERNAL, result["error"]["code"])  # type: ignore[index]
-
     def test_health_endpoint(self):
         conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=5)
         try: