fix: bound heavy gateway operations

bot_bottle/gateway/bounded_http.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import http.server
+import io
 import socket
 import threading
 import time
@@ -14,6 +15,10 @@ class Readable(Protocol):
     def read(self, size: int = -1, /) -> bytes: ...
 
 
+class Writable(Protocol):
+    def write(self, data: bytes, /) -> object: ...
+
+
 @dataclass(frozen=True)
 class BodyReadError(Exception):
     status: int
@@ -30,6 +35,25 @@ def read_declared_body(
     require_length: bool,
 ) -> bytes:
     """Validate and read exactly one declared body under a read deadline."""
+    output = io.BytesIO()
+    copy_declared_body(
+        stream, output, connection, raw_length, maximum=maximum,
+        timeout_seconds=timeout_seconds, require_length=require_length,
+    )
+    return output.getvalue()
+
+
+def copy_declared_body(
+    stream: Readable,
+    output: Writable,
+    connection: socket.socket,
+    raw_length: str | None,
+    *,
+    maximum: int,
+    timeout_seconds: float,
+    require_length: bool,
+) -> int:
+    """Copy one declared body to a sink without retaining it in memory."""
     if raw_length is None:
         if require_length:
             raise BodyReadError(411, "Content-Length required")
@@ -44,7 +68,6 @@ def read_declared_body(
         raise BodyReadError(413, "request body too large")
     previous_timeout = connection.gettimeout()
     deadline = time.monotonic() + timeout_seconds
-    chunks: list[bytes] = []
     remaining = length
     try:
         while remaining:
@@ -55,13 +78,13 @@ def read_declared_body(
             chunk = stream.read(min(remaining, 64 * 1024))
             if not chunk:
                 raise BodyReadError(400, "incomplete request body")
-            chunks.append(chunk)
+            output.write(chunk)
             remaining -= len(chunk)
     except TimeoutError as exc:
         raise BodyReadError(408, "request body read timed out") from exc
     finally:
         connection.settimeout(previous_timeout)
-    return b"".join(chunks)
+    return length
 
 
 class BoundedThreadingHTTPServer(http.server.ThreadingHTTPServer):
@@ -106,4 +129,9 @@ class BoundedThreadingHTTPServer(http.server.ThreadingHTTPServer):
             self._request_slots.release()
 
 
-__all__ = ["BodyReadError", "BoundedThreadingHTTPServer", "read_declared_body"]
+__all__ = [
+    "BodyReadError",
+    "BoundedThreadingHTTPServer",
+    "copy_declared_body",
+    "read_declared_body",
+]

bot_bottle/gateway/git_gate/http_backend.py

@@ -21,6 +21,8 @@ from __future__ import annotations
 import os
 import subprocess
 import sys
+import tempfile
+import threading
 import typing
 from http.server import BaseHTTPRequestHandler
 from pathlib import Path
@@ -30,7 +32,7 @@ from bot_bottle.constants import GIT_GATE_TIMEOUT_SECS, IDENTITY_HEADER
 from bot_bottle.gateway.bounded_http import (
     BodyReadError,
     BoundedThreadingHTTPServer,
-    read_declared_body,
+    copy_declared_body,
 )
 from bot_bottle.gateway.policy_resolver import PolicyResolveError, PolicyResolver
 
@@ -84,6 +86,8 @@ def resolve_sandbox_root(
 MAX_BODY_BYTES = 100 * 1024 * 1024
 REQUEST_BODY_TIMEOUT_SECONDS = 30.0
 MAX_REQUEST_WORKERS = 16
+MAX_BODY_WORKERS = 2
+_BODY_WORK_SLOTS = threading.BoundedSemaphore(MAX_BODY_WORKERS)
 
 
 class GitHttpHandler(BaseHTTPRequestHandler):
@@ -191,31 +195,40 @@ class GitHttpHandler(BaseHTTPRequestHandler):
             value = self.headers.get(header)
             if value:
                 env[variable] = value
-        try:
-            body = read_declared_body(
-                self.rfile,
-                self.connection,
-                self.headers.get("content-length"),
-                maximum=MAX_BODY_BYTES,
-                timeout_seconds=REQUEST_BODY_TIMEOUT_SECONDS,
-                require_length=False,
-            )
-        except BodyReadError as exc:
-            self.send_error(exc.status, exc.message)
+        if not _BODY_WORK_SLOTS.acquire(blocking=False):
+            self.send_error(503, "git request capacity exhausted")
             return
         try:
-            proc = subprocess.run(
-                ["git", "http-backend"],
-                input=body,
-                env=env,
-                capture_output=True,
-                check=False,
-                timeout=GIT_GATE_TIMEOUT_SECS,
-            )
-        except (OSError, subprocess.SubprocessError) as exc:
-            self.log_message("git http-backend unavailable: %s", exc)
-            self.send_error(503, "git backend unavailable")
-            return
+            with tempfile.TemporaryFile() as body:
+                try:
+                    copy_declared_body(
+                        self.rfile,
+                        body,
+                        self.connection,
+                        self.headers.get("content-length"),
+                        maximum=MAX_BODY_BYTES,
+                        timeout_seconds=REQUEST_BODY_TIMEOUT_SECONDS,
+                        require_length=False,
+                    )
+                except BodyReadError as exc:
+                    self.send_error(exc.status, exc.message)
+                    return
+                body.seek(0)
+                try:
+                    proc = subprocess.run(
+                        ["git", "http-backend"],
+                        stdin=body,
+                        env=env,
+                        capture_output=True,
+                        check=False,
+                        timeout=GIT_GATE_TIMEOUT_SECS,
+                    )
+                except (OSError, subprocess.SubprocessError) as exc:
+                    self.log_message("git http-backend unavailable: %s", exc)
+                    self.send_error(503, "git backend unavailable")
+                    return
+        finally:
+            _BODY_WORK_SLOTS.release()
         self._write_cgi_response(proc.stdout)
 
     def _repo_dir(self, sandbox_root: Path, path: str) -> Path | None: