Read fragmented HTTP status lines in the job tests

.github/workflows/ci.yml

@@ -17,6 +17,8 @@ jobs:
       - uses: actions/setup-python@v5
         with:
           python-version: "3.12"
+      - name: Job HTTP harness tests
+        run: python3 -m unittest discover -s scripts -p 'test_job_server_checks.py'
       - uses: dtolnay/rust-toolchain@stable
         with:
           targets: wasm32-wasip1

scripts/job_server_checks.py

@@ -15,6 +15,32 @@
 from job_integration import REPORT_ORDERS, TOKEN, check, workers_running
 
 
+def read_status_line(stream, timeout=10, max_bytes=4096):
+    """Read through CRLF across TCP fragments, within one time and size budget."""
+    deadline = time.monotonic() + timeout
+    previous_timeout = stream.gettimeout()
+    reply = bytearray()
+    try:
+        while len(reply) < max_bytes:
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                raise AssertionError('timed out reading HTTP status line')
+            stream.settimeout(remaining)
+            try:
+                chunk = stream.recv(max_bytes - len(reply))
+            except socket.timeout as error:
+                raise AssertionError('timed out reading HTTP status line') from error
+            if not chunk:
+                raise AssertionError(f'EOF before complete HTTP status line: {bytes(reply)!r}')
+            reply.extend(chunk)
+            end = reply.find(b'\r\n')
+            if end != -1:
+                return reply[:end].decode()
+        raise AssertionError(f'HTTP status line exceeds {max_bytes} bytes')
+    finally:
+        stream.settimeout(previous_timeout)
+
+
 class Server:
     def __init__(self, env, token=TOKEN, flags=None):
         """`flags` replaces the default `--listen 127.0.0.1:<free port>`."""
@@ -168,7 +194,7 @@ def submit(index):
         # Headers only: the refusal must come before the body is read.
         with socket.create_connection(('127.0.0.1', server.port), timeout=10) as raw:
             raw.sendall(f'POST /v1/runs HTTP/1.1\r\nAuthorization: Bearer {TOKEN}\r\nContent-Length: 70000\r\n\r\n'.encode())
-            reply = raw.recv(4096).decode()
+            reply = read_status_line(raw)
         check(reply.startswith('HTTP/1.1 413'), 'an oversized request is refused before it is read', reply)
 
         # The module changes on disk after the policy pinned it.

scripts/test_job_server_checks.py

@@ -0,0 +1,79 @@
+"""Deterministic HTTP harness checks; no Porta binary or OS sandbox required."""
+import socket
+import unittest
+from unittest.mock import Mock, call, patch
+
+from job_server_checks import read_status_line
+
+
+class StatusLineTests(unittest.TestCase):
+    def stream(self, chunks):
+        stream = Mock(spec=socket.socket)
+        stream.gettimeout.return_value = 10
+        stream.recv.side_effect = chunks
+        return stream
+
+    def test_status_split_after_protocol(self):
+        # A single recv saw only this prefix in the Ubuntu job-service CI run.
+        stream = self.stream([b'HTTP/1.1 ', b'413 Payload Too Large\r\n'])
+        self.assertEqual(read_status_line(stream), 'HTTP/1.1 413 Payload Too Large')
+        self.assertEqual(stream.recv.call_count, 2)
+        self.assertEqual(stream.settimeout.call_args, call(10))
+
+    def test_status_and_crlf_split_one_byte_at_a_time(self):
+        line = b'HTTP/1.1 413 Payload Too Large\r\n'
+        stream = self.stream([bytes([byte]) for byte in line])
+        self.assertEqual(read_status_line(stream), line[:-2].decode())
+        self.assertEqual(stream.recv.call_count, len(line))
+
+    def test_stops_at_status_without_waiting_for_headers_body_or_eof(self):
+        stream = self.stream([b'HTTP/1.1 413 Payload Too Large\r\n', socket.timeout()])
+        self.assertEqual(read_status_line(stream), 'HTTP/1.1 413 Payload Too Large')
+        stream.recv.assert_called_once_with(4096)
+
+    def test_ignores_headers_and_body_received_with_status(self):
+        stream = self.stream([b'HTTP/1.1 413 Payload Too Large\r\nContent-Length: 2\r\n\r\n{}'])
+        self.assertEqual(read_status_line(stream), 'HTTP/1.1 413 Payload Too Large')
+
+    def test_unexpected_status_is_preserved(self):
+        stream = self.stream([b'HTTP/1.1 ', b'200 OK\r\n'])
+        self.assertEqual(read_status_line(stream), 'HTTP/1.1 200 OK')
+
+    def test_eof_before_complete_status_fails(self):
+        for prefix in (b'', b'HTTP/1.1 ', b'HTTP/1.1 413 Payload Too Large\r'):
+            with self.subTest(prefix=prefix):
+                stream = self.stream([prefix, b''] if prefix else [b''])
+                with self.assertRaisesRegex(AssertionError, 'EOF before complete HTTP status line'):
+                    read_status_line(stream)
+                self.assertEqual(stream.settimeout.call_args, call(10))
+
+    def test_socket_timeout_fails(self):
+        stream = self.stream([b'HTTP/1.1 ', socket.timeout()])
+        with self.assertRaisesRegex(AssertionError, 'timed out reading HTTP status line'):
+            read_status_line(stream)
+        self.assertEqual(stream.settimeout.call_args, call(10))
+
+    def test_fragments_share_one_deadline(self):
+        stream = self.stream([b'HTTP/1.1 ', b'4'])
+        with patch('job_server_checks.time.monotonic', side_effect=[100, 100, 104, 110]):
+            with self.assertRaisesRegex(AssertionError, 'timed out reading HTTP status line'):
+                read_status_line(stream)
+        self.assertEqual(stream.recv.call_count, 2)
+        self.assertEqual(stream.settimeout.call_args_list, [call(10), call(6), call(10)])
+
+    def test_unterminated_status_cannot_exceed_size_budget(self):
+        stream = self.stream([b'HTTP/1.1 ', b'x' * (4096 - 9)])
+        with self.assertRaisesRegex(AssertionError, 'HTTP status line exceeds 4096 bytes'):
+            read_status_line(stream)
+        self.assertEqual(stream.recv.call_args_list, [call(4096), call(4096 - 9)])
+        self.assertEqual(stream.settimeout.call_args, call(10))
+
+    def test_complete_status_at_size_limit_succeeds(self):
+        line = b'HTTP/1.1 413 Payload Too Large\r\n'
+        stream = self.stream([line])
+        self.assertEqual(read_status_line(stream, max_bytes=len(line)), line[:-2].decode())
+        stream.recv.assert_called_once_with(len(line))
+
+
+if __name__ == '__main__':
+    unittest.main()