diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6edb3da..3211a01 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/scripts/job_server_checks.py b/scripts/job_server_checks.py index 8b7746e..d501e1d 100644 --- a/scripts/job_server_checks.py +++ b/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:`.""" @@ -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. diff --git a/scripts/test_job_server_checks.py b/scripts/test_job_server_checks.py new file mode 100644 index 0000000..a50c57b --- /dev/null +++ b/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()