249 lines
8.4 KiB
Python
249 lines
8.4 KiB
Python
from __future__ import annotations
|
|
|
|
import http.client
|
|
import json
|
|
import os
|
|
import signal
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
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 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 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, 600, 999, "200", 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
|
|
|
|
|
|
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 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)]
|