1079 lines
34 KiB
Python
1079 lines
34 KiB
Python
from __future__ import annotations
|
|
|
|
import http.client
|
|
import json
|
|
import os
|
|
import signal
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import textwrap
|
|
import time
|
|
from collections.abc import Callable
|
|
from contextlib import contextmanager
|
|
from pathlib import Path
|
|
from typing import Iterator
|
|
|
|
import pytest
|
|
|
|
from aios_v2r1 import cli
|
|
from aios_v2r1.ingress_service import (
|
|
MAX_RESPONSE_BYTES,
|
|
MAX_REQUEST_BYTES,
|
|
IngressHTTPServer,
|
|
IngressRequestHandler,
|
|
build_server,
|
|
)
|
|
|
|
|
|
@contextmanager
|
|
def running_server(processor: Callable[[str, bytes], tuple[int, dict[str, object]]] | None = None) -> Iterator[tuple[str, int]]:
|
|
server = build_server("127.0.0.1", 0, processor=processor)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
try:
|
|
host, port = server.server_address
|
|
yield str(host), int(port)
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
thread.join(timeout=5)
|
|
|
|
|
|
def request(
|
|
address: tuple[str, int],
|
|
method: str,
|
|
path: str,
|
|
body: bytes | None = None,
|
|
) -> tuple[int, bytes, dict[str, str]]:
|
|
connection = http.client.HTTPConnection(*address, timeout=5)
|
|
try:
|
|
connection.request(method, path, body=body)
|
|
response = connection.getresponse()
|
|
payload = response.read()
|
|
return response.status, payload, {key.lower(): value for key, value in response.getheaders()}
|
|
finally:
|
|
connection.close()
|
|
|
|
|
|
def raw_request(
|
|
address: tuple[str, int],
|
|
method: bytes,
|
|
path: bytes = b"/health/live",
|
|
) -> bytes:
|
|
request_bytes = (
|
|
method + b" " + path + b" HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n"
|
|
)
|
|
with socket.create_connection(address, timeout=5) as connection:
|
|
connection.sendall(request_bytes)
|
|
chunks: list[bytes] = []
|
|
while True:
|
|
chunk = connection.recv(8192)
|
|
if not chunk:
|
|
return b"".join(chunks)
|
|
chunks.append(chunk)
|
|
|
|
|
|
def raw_exchange(
|
|
address: tuple[str, int],
|
|
request_bytes: bytes,
|
|
*,
|
|
shutdown_write: bool = False,
|
|
receive_timeout: float = 2.0,
|
|
) -> tuple[bytes, bool]:
|
|
chunks: list[bytes] = []
|
|
timed_out = False
|
|
with socket.create_connection(address, timeout=5) as connection:
|
|
connection.settimeout(receive_timeout)
|
|
connection.sendall(request_bytes)
|
|
if shutdown_write:
|
|
connection.shutdown(socket.SHUT_WR)
|
|
while True:
|
|
try:
|
|
chunk = connection.recv(8192)
|
|
except TimeoutError:
|
|
timed_out = True
|
|
break
|
|
except ConnectionResetError:
|
|
break
|
|
if not chunk:
|
|
break
|
|
chunks.append(chunk)
|
|
return b"".join(chunks), timed_out
|
|
|
|
|
|
def parse_raw_response(response: bytes) -> tuple[int, bytes, bytes]:
|
|
headers, body = response.split(b"\r\n\r\n", 1)
|
|
status = int(headers.split(b"\r\n", 1)[0].split(b" ", 2)[1])
|
|
return status, headers, body
|
|
|
|
|
|
def post_request_bytes(headers: bytes, body: bytes = b"") -> bytes:
|
|
return (
|
|
b"POST /v1/ingress HTTP/1.1\r\n"
|
|
b"Host: localhost\r\nConnection: close\r\n"
|
|
+ headers
|
|
+ b"\r\n"
|
|
+ body
|
|
)
|
|
|
|
|
|
def wait_until_listening(process: subprocess.Popen[str], address: tuple[str, int]) -> None:
|
|
deadline = time.monotonic() + 5
|
|
while time.monotonic() < deadline:
|
|
if process.poll() is not None:
|
|
_, stderr = process.communicate()
|
|
pytest.fail(f"server exited before listening: {stderr}")
|
|
try:
|
|
with socket.create_connection(address, timeout=0.1):
|
|
return
|
|
except OSError:
|
|
time.sleep(0.05)
|
|
pytest.fail("server did not start listening")
|
|
|
|
|
|
def test_live_and_unconfigured_readiness_are_exact_and_bounded() -> None:
|
|
with running_server() as address:
|
|
live = request(address, "GET", "/health/live")
|
|
ready = request(address, "GET", "/health/ready")
|
|
|
|
assert live[:2] == (200, b'{"status":"LIVE"}')
|
|
assert ready[:2] == (503, b'{"status":"SERVICE_NOT_CONFIGURED"}')
|
|
assert live[2]["content-type"] == "application/json"
|
|
assert ready[2]["content-length"] == str(len(ready[1]))
|
|
assert len(live[1]) <= 4096
|
|
assert len(ready[1]) <= 4096
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"route",
|
|
["/v1/ingress", "/v1/receipts/status", "/v1/outbound-receipts"],
|
|
)
|
|
def test_mutation_routes_fail_closed_without_reading_or_mutating(route: str) -> None:
|
|
with running_server() as address:
|
|
status, payload, _ = request(
|
|
address,
|
|
"POST",
|
|
route,
|
|
b'{"sender":"do-not-log","body":"do-not-log"}',
|
|
)
|
|
assert status == 503
|
|
assert payload == b'{"status":"SERVICE_NOT_CONFIGURED"}'
|
|
|
|
|
|
def test_unknown_routes_and_wrong_methods_use_bounded_sanitized_codes() -> None:
|
|
with running_server() as address:
|
|
missing = request(address, "GET", "/missing")
|
|
wrong_method = request(address, "GET", "/v1/ingress")
|
|
unsupported = request(address, "PUT", "/health/live")
|
|
|
|
assert missing[:2] == (404, b'{"status":"NOT_FOUND"}')
|
|
assert wrong_method[:2] == (405, b'{"status":"METHOD_NOT_ALLOWED"}')
|
|
assert unsupported[:2] == (405, b'{"status":"METHOD_NOT_ALLOWED"}')
|
|
assert all(len(result[1]) <= 4096 for result in (missing, wrong_method, unsupported))
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("method", "expected_status"),
|
|
[
|
|
(b"METHOD_SECRET_REFLECTION_TOKEN", b"405"),
|
|
(b"METHOD_SECRET_REFLECTION_TOKEN" + b"X" * 66_000, b"414"),
|
|
],
|
|
ids=["arbitrary", "overlong"],
|
|
)
|
|
def test_arbitrary_and_overlong_methods_use_bounded_json_without_reflection(
|
|
method: bytes, expected_status: bytes
|
|
) -> None:
|
|
with running_server() as address:
|
|
response = raw_request(address, method)
|
|
|
|
headers, body = response.split(b"\r\n\r\n", 1)
|
|
status_line = headers.split(b"\r\n", 1)[0]
|
|
assert status_line.split(b" ", 2)[1] == expected_status
|
|
assert b"Content-Type: application/json" in headers
|
|
assert len(body) <= 4096
|
|
assert b"METHOD_SECRET_REFLECTION_TOKEN" not in response
|
|
assert b"<html" not in response.lower()
|
|
|
|
|
|
@pytest.mark.parametrize("invalid_status", [True, False, 99, 100, 101, 199, 204, 205, 304, 600, 999, "200", 200.0, None])
|
|
def test_invalid_processor_status_is_sanitized_before_writing_status_line(
|
|
invalid_status: object,
|
|
) -> None:
|
|
def processor(route: str, body: bytes) -> tuple[int, dict[str, object]]:
|
|
return invalid_status, {"status": "MUST_NOT_ESCAPE"} # type: ignore[return-value]
|
|
|
|
with running_server(processor) as address:
|
|
response = raw_request(address, b"POST", b"/v1/ingress")
|
|
|
|
headers, body = response.split(b"\r\n\r\n", 1)
|
|
assert headers.split(b"\r\n", 1)[0].startswith(b"HTTP/1.0 500 ")
|
|
assert body == b'{"status":"INTERNAL_ERROR"}'
|
|
assert b"MUST_NOT_ESCAPE" not in response
|
|
|
|
|
|
def test_processor_is_injected_and_exceptions_are_not_disclosed(capsys: pytest.CaptureFixture[str]) -> None:
|
|
secret = "sender-secret-body-token"
|
|
|
|
def exploding_processor(route: str, body: bytes) -> tuple[int, dict[str, object]]:
|
|
assert route == "/v1/ingress"
|
|
assert json.loads(body) == {"value": secret}
|
|
raise RuntimeError(secret)
|
|
|
|
with running_server(exploding_processor) as address:
|
|
ready = request(address, "GET", "/health/ready")
|
|
failed = request(
|
|
address,
|
|
"POST",
|
|
"/v1/ingress",
|
|
json.dumps({"value": secret}).encode(),
|
|
)
|
|
|
|
captured = capsys.readouterr()
|
|
assert ready[:2] == (200, b'{"status":"READY"}')
|
|
assert failed[:2] == (500, b'{"status":"INTERNAL_ERROR"}')
|
|
assert secret not in failed[1].decode()
|
|
assert secret not in captured.out
|
|
assert secret not in captured.err
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("headers", "body"),
|
|
[
|
|
(b"Transfer-Encoding: secret-framing-token\r\n", b"secret-body"),
|
|
(b"Content-Length: 0\r\nContent-Length: 0\r\n", b""),
|
|
(b"Content-Length: 1\r\nContent-Length: 2\r\n", b"x"),
|
|
(b"Content-Length: invalid-secret-length\r\n", b""),
|
|
(b"Content-Length: -1\r\n", b""),
|
|
(f"Content-Length: {MAX_REQUEST_BYTES + 1}\r\n".encode(), b""),
|
|
(b"Content-Length: 0 \r\n", b""),
|
|
(b"Content-Length:\t0\r\n", b""),
|
|
],
|
|
ids=[
|
|
"transfer-encoding",
|
|
"duplicate-same",
|
|
"duplicate-conflicting",
|
|
"invalid",
|
|
"negative",
|
|
"oversize",
|
|
"trailing-ows",
|
|
"leading-ows",
|
|
],
|
|
)
|
|
def test_invalid_post_framing_is_bounded_and_never_calls_processor(
|
|
headers: bytes,
|
|
body: bytes,
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
calls: list[bytes] = []
|
|
|
|
def processor(route: str, payload: bytes) -> tuple[int, dict[str, object]]:
|
|
calls.append(payload)
|
|
return 201, {"status": "MUTATED"}
|
|
|
|
with running_server(processor) as address:
|
|
response, timed_out = raw_exchange(
|
|
address,
|
|
post_request_bytes(headers, body),
|
|
)
|
|
|
|
status, response_headers, response_body = parse_raw_response(response)
|
|
captured = capsys.readouterr()
|
|
assert not timed_out
|
|
assert 400 <= status <= 499
|
|
assert b"Content-Type: application/json" in response_headers
|
|
assert len(response_body) <= 4096
|
|
assert b"secret-framing-token" not in response
|
|
assert b"secret-body" not in response
|
|
assert b"invalid-secret-length" not in response
|
|
assert calls == []
|
|
assert "Traceback" not in captured.out
|
|
assert "Traceback" not in captured.err
|
|
|
|
|
|
def test_short_eof_body_is_rejected_before_processor() -> None:
|
|
calls: list[bytes] = []
|
|
|
|
def processor(route: str, payload: bytes) -> tuple[int, dict[str, object]]:
|
|
calls.append(payload)
|
|
return 201, {"status": "MUTATED"}
|
|
|
|
with running_server(processor) as address:
|
|
response, timed_out = raw_exchange(
|
|
address,
|
|
post_request_bytes(b"Content-Length: 10\r\n", b"abc"),
|
|
shutdown_write=True,
|
|
)
|
|
|
|
status, _, body = parse_raw_response(response)
|
|
assert not timed_out
|
|
assert 400 <= status <= 499
|
|
assert len(body) <= 4096
|
|
assert calls == []
|
|
|
|
|
|
def test_partial_body_timeout_is_bounded_and_rejected_before_processor() -> None:
|
|
processor_called = threading.Event()
|
|
|
|
def processor(route: str, payload: bytes) -> tuple[int, dict[str, object]]:
|
|
processor_called.set()
|
|
return 201, {"status": "MUTATED"}
|
|
|
|
with running_server(processor) as address:
|
|
started = time.monotonic()
|
|
response, timed_out = raw_exchange(
|
|
address,
|
|
post_request_bytes(b"Content-Length: 10\r\n", b"abc"),
|
|
receive_timeout=1.5,
|
|
)
|
|
elapsed = time.monotonic() - started
|
|
called_after_client_close = processor_called.wait(timeout=0.5)
|
|
|
|
status, _, body = parse_raw_response(response)
|
|
assert not timed_out
|
|
assert elapsed < 1.5
|
|
assert status == 408
|
|
assert len(body) <= 4096
|
|
assert not called_after_client_close
|
|
|
|
|
|
def test_server_close_drains_partial_body_handler_and_prevents_late_processor(
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
processor_called = threading.Event()
|
|
|
|
def processor(route: str, payload: bytes) -> tuple[int, dict[str, object]]:
|
|
processor_called.set()
|
|
return 201, {"status": "MUTATED"}
|
|
|
|
server = build_server("127.0.0.1", 0, processor=processor)
|
|
server_thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
server_thread.start()
|
|
host, port = server.server_address
|
|
client = socket.create_connection((str(host), int(port)), timeout=5)
|
|
try:
|
|
client.sendall(post_request_bytes(b"Content-Length: 10\r\n", b"abc"))
|
|
time.sleep(0.1)
|
|
started = time.monotonic()
|
|
server.shutdown()
|
|
server.server_close()
|
|
close_elapsed = time.monotonic() - started
|
|
finally:
|
|
client.close()
|
|
server_thread.join(timeout=5)
|
|
|
|
called_after_close = processor_called.wait(timeout=0.5)
|
|
captured = capsys.readouterr()
|
|
assert server.daemon_threads is False
|
|
assert server.block_on_close is True
|
|
assert close_elapsed < 3
|
|
assert not server_thread.is_alive()
|
|
assert not called_after_close
|
|
assert "Traceback" not in captured.out
|
|
assert "Traceback" not in captured.err
|
|
assert "Fatal Python error" not in captured.err
|
|
restarted = build_server(str(host), int(port))
|
|
restarted.server_close()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"target",
|
|
[
|
|
b"http://[::1/v1/ingress",
|
|
b"http://127.0.0.1:bad/v1/ingress",
|
|
b"relative-target",
|
|
b"/v1/ingress%ZZ",
|
|
],
|
|
ids=[
|
|
"invalid-ipv6-bracket",
|
|
"invalid-absolute-port",
|
|
"invalid-origin-form",
|
|
"invalid-percent-escape",
|
|
],
|
|
)
|
|
def test_malformed_request_targets_return_sanitized_400_without_processor(
|
|
target: bytes,
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
calls: list[bytes] = []
|
|
|
|
def processor(route: str, payload: bytes) -> tuple[int, dict[str, object]]:
|
|
calls.append(payload)
|
|
return 201, {"status": "MUTATED"}
|
|
|
|
request_bytes = (
|
|
b"POST "
|
|
+ target
|
|
+ b" HTTP/1.1\r\nHost: localhost\r\n"
|
|
+ b"Content-Length: 0\r\nConnection: close\r\n\r\n"
|
|
)
|
|
with running_server(processor) as address:
|
|
response, timed_out = raw_exchange(address, request_bytes)
|
|
|
|
status, headers, body = parse_raw_response(response)
|
|
captured = capsys.readouterr()
|
|
assert not timed_out
|
|
assert status == 400
|
|
assert b"Content-Type: application/json" in headers
|
|
assert len(body) <= 4096
|
|
assert target not in response
|
|
assert calls == []
|
|
assert "Traceback" not in captured.out
|
|
assert "Traceback" not in captured.err
|
|
|
|
|
|
def test_server_handle_error_never_prints_payload_traceback(
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
secret = "unexpected-handler-secret"
|
|
|
|
class ExplodingHandler(IngressRequestHandler):
|
|
def do_GET(self) -> None:
|
|
raise RuntimeError(secret)
|
|
|
|
server = IngressHTTPServer(("127.0.0.1", 0), ExplodingHandler, None)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
try:
|
|
host, port = server.server_address
|
|
raw_exchange(
|
|
(str(host), int(port)),
|
|
b"GET /health/live HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n",
|
|
)
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
thread.join(timeout=5)
|
|
|
|
captured = capsys.readouterr()
|
|
assert secret not in captured.out
|
|
assert secret not in captured.err
|
|
assert "Traceback" not in captured.out
|
|
assert "Traceback" not in captured.err
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"malformed_header",
|
|
[
|
|
b"Bad Name: value\r\n",
|
|
b"Bad\tName: value\r\n",
|
|
b"Trailing-Name : value\r\n",
|
|
b" X-Leading: value\r\n",
|
|
b"\tX-Leading: value\r\n",
|
|
b":authority: localhost\r\n",
|
|
b"X-Obs-Fold: first\r\n continuation\r\n",
|
|
b"X-Obs-Fold: first\r\n\tcontinuation\r\n",
|
|
b"X-Control: value\x00secret\r\n",
|
|
b"X-Carriage: value\rsecret\r\n",
|
|
b"X-Bare-LF: value\n",
|
|
b"X-\xff: value\r\n",
|
|
],
|
|
ids=[
|
|
"space-in-name",
|
|
"tab-in-name",
|
|
"whitespace-before-colon",
|
|
"leading-space",
|
|
"leading-tab",
|
|
"pseudo-header",
|
|
"obs-fold-space",
|
|
"obs-fold-tab",
|
|
"nul-in-value",
|
|
"cr-in-value",
|
|
"bare-lf",
|
|
"non-ascii-name",
|
|
],
|
|
)
|
|
def test_malformed_raw_header_grammar_is_rejected_before_processor(
|
|
malformed_header: bytes,
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
calls: list[bytes] = []
|
|
|
|
def processor(route: str, payload: bytes) -> tuple[int, dict[str, object]]:
|
|
calls.append(payload)
|
|
return 201, {"status": "MUTATED"}
|
|
|
|
request_bytes = (
|
|
b"POST /v1/ingress HTTP/1.1\r\n"
|
|
b"Host: localhost\r\n"
|
|
+ malformed_header
|
|
+ b"Content-Length: 0\r\nConnection: close\r\n\r\n"
|
|
)
|
|
with running_server(processor) as address:
|
|
response, timed_out = raw_exchange(address, request_bytes)
|
|
|
|
assert not timed_out
|
|
assert response
|
|
status, headers, body = parse_raw_response(response)
|
|
captured = capsys.readouterr()
|
|
assert status == 400
|
|
assert b"Content-Type: application/json" in headers
|
|
assert json.loads(body) == {"status": "BAD_REQUEST"}
|
|
assert len(body) <= MAX_RESPONSE_BYTES
|
|
assert b"secret" not in response
|
|
assert calls == []
|
|
assert "Traceback" not in captured.out
|
|
assert "Traceback" not in captured.err
|
|
|
|
|
|
def trickle_request_until_response(
|
|
address: tuple[str, int],
|
|
prefix: bytes,
|
|
*,
|
|
receive_timeout: float = 1.8,
|
|
) -> tuple[bytes, bool, float]:
|
|
connection = socket.create_connection(address, timeout=5)
|
|
stop = threading.Event()
|
|
|
|
def sender() -> None:
|
|
while not stop.wait(0.1):
|
|
try:
|
|
connection.sendall(b"x")
|
|
except OSError:
|
|
return
|
|
|
|
connection.sendall(prefix)
|
|
sender_thread = threading.Thread(target=sender, daemon=True)
|
|
sender_thread.start()
|
|
connection.settimeout(receive_timeout)
|
|
chunks: list[bytes] = []
|
|
timed_out = False
|
|
started = time.monotonic()
|
|
try:
|
|
while True:
|
|
try:
|
|
chunk = connection.recv(8192)
|
|
except TimeoutError:
|
|
timed_out = True
|
|
break
|
|
except ConnectionResetError:
|
|
break
|
|
if not chunk:
|
|
break
|
|
chunks.append(chunk)
|
|
finally:
|
|
elapsed = time.monotonic() - started
|
|
stop.set()
|
|
try:
|
|
connection.shutdown(socket.SHUT_RDWR)
|
|
except OSError:
|
|
pass
|
|
connection.close()
|
|
sender_thread.join(timeout=1)
|
|
return b"".join(chunks), timed_out, elapsed
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"prefix",
|
|
[
|
|
(
|
|
b"POST /v1/ingress HTTP/1.1\r\n"
|
|
b"Host: localhost\r\nX-Slow-Header: "
|
|
),
|
|
(
|
|
b"POST /v1/ingress HTTP/1.1\r\n"
|
|
b"Host: localhost\r\nContent-Length: 1000\r\n"
|
|
b"Connection: close\r\n\r\nx"
|
|
),
|
|
],
|
|
ids=["slow-header", "slow-body"],
|
|
)
|
|
def test_absolute_network_deadline_cannot_be_reset_by_slow_trickle(
|
|
prefix: bytes,
|
|
) -> None:
|
|
processor_called = threading.Event()
|
|
|
|
def processor(route: str, payload: bytes) -> tuple[int, dict[str, object]]:
|
|
processor_called.set()
|
|
return 201, {"status": "MUTATED"}
|
|
|
|
with running_server(processor) as address:
|
|
response, timed_out, elapsed = trickle_request_until_response(address, prefix)
|
|
called_after_cleanup = processor_called.wait(timeout=0.4)
|
|
|
|
assert not timed_out
|
|
assert elapsed < 1.8
|
|
assert response
|
|
status, _, body = parse_raw_response(response)
|
|
assert status == 408
|
|
assert json.loads(body) == {"status": "REQUEST_TIMEOUT"}
|
|
assert len(body) <= MAX_RESPONSE_BYTES
|
|
assert not called_after_cleanup
|
|
|
|
|
|
def test_shutdown_drains_several_slow_clients_without_late_work_or_threads(
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
processor_called = threading.Event()
|
|
|
|
def processor(route: str, payload: bytes) -> tuple[int, dict[str, object]]:
|
|
processor_called.set()
|
|
return 201, {"status": "MUTATED"}
|
|
|
|
before_non_daemon = {id(thread) for thread in threading.enumerate() if not thread.daemon}
|
|
server = build_server("127.0.0.1", 0, processor=processor)
|
|
server_thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
server_thread.start()
|
|
host, port = server.server_address
|
|
clients: list[socket.socket] = []
|
|
stops: list[threading.Event] = []
|
|
senders: list[threading.Thread] = []
|
|
|
|
def start_client(prefix: bytes) -> None:
|
|
client = socket.create_connection((str(host), int(port)), timeout=5)
|
|
client.sendall(prefix)
|
|
stop = threading.Event()
|
|
|
|
def sender() -> None:
|
|
while not stop.wait(0.1):
|
|
try:
|
|
client.sendall(b"x")
|
|
except OSError:
|
|
return
|
|
|
|
sender_thread = threading.Thread(target=sender, daemon=True)
|
|
sender_thread.start()
|
|
clients.append(client)
|
|
stops.append(stop)
|
|
senders.append(sender_thread)
|
|
|
|
header_prefix = (
|
|
b"POST /v1/ingress HTTP/1.1\r\nHost: localhost\r\nX-Slow: "
|
|
)
|
|
body_prefix = (
|
|
b"POST /v1/ingress HTTP/1.1\r\nHost: localhost\r\n"
|
|
b"Content-Length: 1000\r\nConnection: close\r\n\r\nx"
|
|
)
|
|
for index in range(6):
|
|
start_client(header_prefix if index % 2 == 0 else body_prefix)
|
|
time.sleep(0.15)
|
|
|
|
closed = threading.Event()
|
|
close_elapsed: list[float] = []
|
|
|
|
def close_server() -> None:
|
|
started = time.monotonic()
|
|
try:
|
|
server.shutdown()
|
|
server.server_close()
|
|
finally:
|
|
close_elapsed.append(time.monotonic() - started)
|
|
closed.set()
|
|
|
|
closer = threading.Thread(target=close_server)
|
|
closer.start()
|
|
closed_within_bound = closed.wait(timeout=2.5)
|
|
for stop in stops:
|
|
stop.set()
|
|
for client in clients:
|
|
try:
|
|
client.shutdown(socket.SHUT_RDWR)
|
|
except OSError:
|
|
pass
|
|
client.close()
|
|
for sender_thread in senders:
|
|
sender_thread.join(timeout=1)
|
|
closer.join(timeout=5)
|
|
server_thread.join(timeout=5)
|
|
|
|
called_after_close = processor_called.wait(timeout=0.4)
|
|
deadline = time.monotonic() + 1
|
|
leaked_non_daemon: list[threading.Thread] = []
|
|
while time.monotonic() < deadline:
|
|
leaked_non_daemon = [
|
|
thread
|
|
for thread in threading.enumerate()
|
|
if not thread.daemon
|
|
and id(thread) not in before_non_daemon
|
|
and thread.is_alive()
|
|
]
|
|
if not leaked_non_daemon:
|
|
break
|
|
time.sleep(0.02)
|
|
captured = capsys.readouterr()
|
|
assert closed_within_bound
|
|
assert close_elapsed and close_elapsed[0] < 2.5
|
|
assert not server_thread.is_alive()
|
|
assert not closer.is_alive()
|
|
assert not leaked_non_daemon
|
|
assert not called_after_close
|
|
assert "Traceback" not in captured.out
|
|
assert "Traceback" not in captured.err
|
|
assert "Fatal Python error" not in captured.err
|
|
restarted = build_server(str(host), int(port))
|
|
restarted.server_close()
|
|
|
|
|
|
def test_completed_framing_cancels_deadline_before_slow_processor() -> None:
|
|
def processor(route: str, payload: bytes) -> tuple[int, dict[str, object]]:
|
|
time.sleep(1.2)
|
|
return 201, {"status": "MUTATED"}
|
|
|
|
with running_server(processor) as address:
|
|
status, body, _ = request(address, "POST", "/v1/ingress", b"")
|
|
|
|
assert status == 201
|
|
assert json.loads(body) == {"status": "MUTATED"}
|
|
assert not [
|
|
thread
|
|
for thread in threading.enumerate()
|
|
if thread.name.startswith("aios-ingress-read-deadline")
|
|
and thread.is_alive()
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"payload_factory",
|
|
[
|
|
pytest.param(lambda: [], id="list"),
|
|
pytest.param(lambda: "not-an-object", id="string"),
|
|
pytest.param(lambda: None, id="null"),
|
|
pytest.param(lambda: {"value": float("nan")}, id="nan"),
|
|
pytest.param(lambda: {"value": float("inf")}, id="infinity"),
|
|
pytest.param(lambda: {"value": float("-inf")}, id="negative-infinity"),
|
|
pytest.param(lambda: {"value": object()}, id="nonserializable"),
|
|
pytest.param(
|
|
lambda: {"value": "secret-oversize-" + "x" * MAX_RESPONSE_BYTES},
|
|
id="oversize",
|
|
),
|
|
],
|
|
)
|
|
def test_invalid_processor_payload_is_fixed_bounded_json_500(
|
|
payload_factory: Callable[[], object],
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
payload = payload_factory()
|
|
|
|
def processor(route: str, body: bytes) -> tuple[int, dict[str, object]]:
|
|
return 201, payload # type: ignore[return-value]
|
|
|
|
with running_server(processor) as address:
|
|
response = raw_request(address, b"POST", b"/v1/ingress")
|
|
|
|
status, headers, body = parse_raw_response(response)
|
|
captured = capsys.readouterr()
|
|
assert status == 500
|
|
assert b"Content-Type: application/json" in headers
|
|
assert body == b'{"status":"INTERNAL_ERROR"}'
|
|
assert json.loads(body) == {"status": "INTERNAL_ERROR"}
|
|
assert len(body) <= MAX_RESPONSE_BYTES
|
|
assert b"secret-oversize" not in response
|
|
assert "Traceback" not in captured.out
|
|
assert "Traceback" not in captured.err
|
|
|
|
|
|
def test_recursive_processor_payload_uses_nonrecursive_fixed_500() -> None:
|
|
recursive: dict[str, object] = {}
|
|
recursive["self"] = recursive
|
|
|
|
def processor(route: str, body: bytes) -> tuple[int, dict[str, object]]:
|
|
return 201, recursive
|
|
|
|
with running_server(processor) as address:
|
|
status, body, _ = request(address, "POST", "/v1/ingress", b"")
|
|
|
|
assert status == 500
|
|
assert body == b'{"status":"INTERNAL_ERROR"}'
|
|
assert json.loads(body) == {"status": "INTERNAL_ERROR"}
|
|
|
|
|
|
def test_server_rejects_non_loopback_bind() -> None:
|
|
with pytest.raises(ValueError, match="loopback"):
|
|
build_server("0.0.0.0", 0)
|
|
|
|
|
|
def test_af_inet_server_rejects_ipv6_loopback_explicitly() -> None:
|
|
with pytest.raises(ValueError, match="IPv4 loopback"):
|
|
build_server("::1", 0)
|
|
|
|
|
|
|
|
def run_pre_header_sigint_probe(
|
|
timeout: float = 1.2,
|
|
) -> tuple[int, str, str, float, bool, int]:
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as reservation:
|
|
reservation.bind(("127.0.0.1", 0))
|
|
port = int(reservation.getsockname()[1])
|
|
|
|
script = textwrap.dedent(
|
|
"""
|
|
import os
|
|
import signal
|
|
import socket
|
|
import sys
|
|
import threading
|
|
import time
|
|
|
|
from aios_v2r1.ingress_service import (
|
|
IngressHTTPServer,
|
|
IngressRequestHandler,
|
|
)
|
|
|
|
accepted_before_setup = threading.Event()
|
|
|
|
class PreHeaderReadinessHandler(IngressRequestHandler):
|
|
def setup(self):
|
|
accepted_before_setup.set()
|
|
try:
|
|
self.request.recv(1)
|
|
except OSError:
|
|
pass
|
|
super().setup()
|
|
|
|
server = IngressHTTPServer(
|
|
("127.0.0.1", int(sys.argv[1])),
|
|
PreHeaderReadinessHandler,
|
|
None,
|
|
)
|
|
|
|
def connect_and_interrupt():
|
|
client = socket.create_connection(server.server_address, timeout=2)
|
|
if not accepted_before_setup.wait(timeout=2):
|
|
os._exit(71)
|
|
os.kill(os.getpid(), signal.SIGINT)
|
|
time.sleep(10)
|
|
client.close()
|
|
|
|
trigger = threading.Thread(target=connect_and_interrupt, daemon=True)
|
|
trigger.start()
|
|
try:
|
|
server.serve_forever(poll_interval=0.01)
|
|
except KeyboardInterrupt:
|
|
pass
|
|
finally:
|
|
server.server_close()
|
|
"""
|
|
)
|
|
env = os.environ.copy()
|
|
package_root = str(Path(__file__).resolve().parents[2] / "active-runtime" / "app")
|
|
env["PYTHONPATH"] = package_root + os.pathsep + env.get("PYTHONPATH", "")
|
|
process = subprocess.Popen(
|
|
[sys.executable, "-c", script, str(port)],
|
|
env=env,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
)
|
|
started = time.monotonic()
|
|
timed_out = False
|
|
try:
|
|
stdout, stderr = process.communicate(timeout=timeout)
|
|
except subprocess.TimeoutExpired:
|
|
timed_out = True
|
|
process.kill()
|
|
stdout, stderr = process.communicate(timeout=2)
|
|
return (
|
|
int(process.returncode),
|
|
stdout,
|
|
stderr,
|
|
time.monotonic() - started,
|
|
timed_out,
|
|
port,
|
|
)
|
|
|
|
|
|
def test_sigint_closes_pre_header_readiness_socket_in_25_bounded_runs() -> None:
|
|
for iteration in range(25):
|
|
returncode, stdout, stderr, elapsed, timed_out, port = (
|
|
run_pre_header_sigint_probe()
|
|
)
|
|
assert not timed_out, f"pre-header SIGINT hung on iteration {iteration}"
|
|
assert elapsed < 1.2
|
|
assert returncode == 0
|
|
assert "Traceback" not in stdout
|
|
assert "Traceback" not in stderr
|
|
assert "Fatal Python error" not in stderr
|
|
restarted = build_server("127.0.0.1", port)
|
|
restarted.server_close()
|
|
|
|
|
|
def test_server_close_closes_socket_accepted_before_handler_setup() -> None:
|
|
server = build_server("127.0.0.1", 0)
|
|
client = socket.create_connection(server.server_address, timeout=2)
|
|
accepted, _ = server.get_request()
|
|
client.settimeout(0.4)
|
|
try:
|
|
server.server_close()
|
|
server.server_close()
|
|
try:
|
|
remaining = client.recv(1)
|
|
except ConnectionResetError:
|
|
remaining = b""
|
|
assert remaining == b""
|
|
finally:
|
|
accepted.close()
|
|
client.close()
|
|
server.server_close()
|
|
|
|
|
|
def test_process_request_rejects_socket_after_server_closing() -> None:
|
|
handler_started = threading.Event()
|
|
|
|
class RecordingHandler:
|
|
def __init__(self, *args: object) -> None:
|
|
handler_started.set()
|
|
|
|
server = IngressHTTPServer(
|
|
("127.0.0.1", 0),
|
|
RecordingHandler, # type: ignore[arg-type]
|
|
None,
|
|
)
|
|
server.server_close()
|
|
request_socket, peer = socket.socketpair()
|
|
peer.settimeout(0.4)
|
|
try:
|
|
server.process_request(request_socket, ("127.0.0.1", 1))
|
|
try:
|
|
remaining = peer.recv(1)
|
|
except ConnectionResetError:
|
|
remaining = b""
|
|
assert remaining == b""
|
|
assert not handler_started.wait(timeout=0.1)
|
|
finally:
|
|
request_socket.close()
|
|
peer.close()
|
|
server.server_close()
|
|
|
|
|
|
|
|
def test_accept_and_server_close_race_is_bounded_in_25_runs() -> None:
|
|
errors: list[BaseException] = []
|
|
for _ in range(25):
|
|
class RecordingHandler:
|
|
def __init__(self, *args: object) -> None:
|
|
return
|
|
|
|
server = IngressHTTPServer(
|
|
("127.0.0.1", 0),
|
|
RecordingHandler, # type: ignore[arg-type]
|
|
None,
|
|
)
|
|
client = socket.create_connection(server.server_address, timeout=2)
|
|
barrier = threading.Barrier(3)
|
|
|
|
def accept_and_process() -> None:
|
|
try:
|
|
barrier.wait(timeout=1)
|
|
request_socket, client_address = server.get_request()
|
|
server.process_request(request_socket, client_address)
|
|
except OSError:
|
|
return
|
|
except BaseException as exc:
|
|
errors.append(exc)
|
|
|
|
def close_during_accept() -> None:
|
|
try:
|
|
barrier.wait(timeout=1)
|
|
server.server_close()
|
|
except BaseException as exc:
|
|
errors.append(exc)
|
|
|
|
accepter = threading.Thread(target=accept_and_process)
|
|
closer = threading.Thread(target=close_during_accept)
|
|
accepter.start()
|
|
closer.start()
|
|
barrier.wait(timeout=1)
|
|
accepter.join(timeout=1)
|
|
closer.join(timeout=1)
|
|
client.close()
|
|
server.server_close()
|
|
assert not accepter.is_alive()
|
|
assert not closer.is_alive()
|
|
assert not errors
|
|
|
|
def test_server_close_is_idempotent_with_idle_and_partial_clients(
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
server = build_server("127.0.0.1", 0)
|
|
server_thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
server_thread.start()
|
|
clients: list[socket.socket] = []
|
|
payloads = [
|
|
None,
|
|
None,
|
|
b"GET /health/live HTTP/1.1\r\nHost: localhost\r\n",
|
|
b"POST /v1/ingress HTTP/1.1\r\nContent-Length: 10\r\n\r\nx",
|
|
]
|
|
for payload in payloads:
|
|
client = socket.create_connection(server.server_address, timeout=2)
|
|
if payload is not None:
|
|
client.sendall(payload)
|
|
clients.append(client)
|
|
time.sleep(0.05)
|
|
server.shutdown()
|
|
|
|
errors: list[BaseException] = []
|
|
|
|
def close_server() -> None:
|
|
try:
|
|
server.server_close()
|
|
except BaseException as exc:
|
|
errors.append(exc)
|
|
|
|
closers = [threading.Thread(target=close_server) for _ in range(4)]
|
|
for closer in closers:
|
|
closer.start()
|
|
for closer in closers:
|
|
closer.join(timeout=2)
|
|
server_thread.join(timeout=2)
|
|
for client in clients:
|
|
client.close()
|
|
captured = capsys.readouterr()
|
|
assert not errors
|
|
assert not [closer for closer in closers if closer.is_alive()]
|
|
assert not server_thread.is_alive()
|
|
assert "Traceback" not in captured.out
|
|
assert "Traceback" not in captured.err
|
|
assert "Fatal Python error" not in captured.err
|
|
|
|
def test_sigint_shutdown_exits_zero_without_traceback_and_releases_port() -> None:
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as reservation:
|
|
reservation.bind(("127.0.0.1", 0))
|
|
port = int(reservation.getsockname()[1])
|
|
|
|
env = os.environ.copy()
|
|
package_root = str(Path(__file__).resolve().parents[2] / "active-runtime" / "app")
|
|
env["PYTHONPATH"] = package_root + os.pathsep + env.get("PYTHONPATH", "")
|
|
process = subprocess.Popen(
|
|
[
|
|
sys.executable,
|
|
"-m",
|
|
"aios_v2r1.cli",
|
|
"serve",
|
|
"--host",
|
|
"127.0.0.1",
|
|
"--port",
|
|
str(port),
|
|
],
|
|
env=env,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
)
|
|
try:
|
|
wait_until_listening(process, ("127.0.0.1", port))
|
|
process.send_signal(signal.SIGINT)
|
|
stdout, stderr = process.communicate(timeout=5)
|
|
finally:
|
|
if process.poll() is None:
|
|
process.kill()
|
|
process.communicate()
|
|
|
|
assert process.returncode == 0
|
|
assert "Traceback" not in stdout
|
|
assert "Traceback" not in stderr
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as rebound:
|
|
rebound.bind(("127.0.0.1", port))
|
|
|
|
|
|
def test_cli_preserves_registry_commands_and_dispatches_top_level_serve(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
parsed = cli.build_parser().parse_args(["registry.sqlite", "status", "tenant-a"])
|
|
assert parsed.registry == "registry.sqlite"
|
|
assert parsed.command == "status"
|
|
|
|
called: list[tuple[str, int]] = []
|
|
monkeypatch.setattr(cli, "serve", lambda host, port: called.append((host, port)))
|
|
assert cli.main(["serve", "--host", "127.0.0.1", "--port", "19001"]) == 0
|
|
assert called == [("127.0.0.1", 19001)]
|