312 lines
13 KiB
Python
312 lines
13 KiB
Python
"""Adversarial HTTP boundary tests for the loopback HUX foundation service."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import socket
|
|
import sys
|
|
import threading
|
|
from http.client import HTTPConnection
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
FOUNDATION = ROOT / "dockerfiles" / "hermes-hux-foundation"
|
|
if str(FOUNDATION) not in sys.path:
|
|
sys.path.insert(0, str(FOUNDATION))
|
|
|
|
from hux import contracts, errors, flags, identity # noqa: E402
|
|
from hux import http as hux_http # noqa: E402
|
|
from hux.http import DEFAULT_REQUEST_TIMEOUT_SECONDS, RateLimiter, Router, serve # noqa: E402
|
|
from hux.server import build_router # noqa: E402
|
|
|
|
ALL_ON = ",".join(card["flag"] for card in contracts.load_flags()["cards"])
|
|
HEADERS = {
|
|
"X-Hermes-Tenant-Identity": "slot-3",
|
|
"X-Hux-Subject": "usr_0123456789abcdef",
|
|
"X-Hux-Surface": "chat",
|
|
"X-Hux-Relay-Key": "rk",
|
|
}
|
|
|
|
|
|
def _router(tmp_path: Path, **extra: str) -> Router:
|
|
return build_router(tmp_path, {"HUX_FLAGS": ALL_ON, "HUX_ROUTER_KEY": "rk", **extra})
|
|
|
|
|
|
def _start(router: Router):
|
|
server = serve(router, "127.0.0.1", 0)
|
|
threading.Thread(target=server.serve_forever, daemon=True).start()
|
|
return server
|
|
|
|
|
|
def test_each_trust_uses_its_own_0400_key_file(tmp_path):
|
|
"""Router, relay and worker credentials are distinct files; weak permissions fail closed."""
|
|
files = {}
|
|
for trust in ("router", "relay", "worker"):
|
|
path = tmp_path / f"{trust}-key"
|
|
path.write_text(f"{trust}-secret\n")
|
|
path.chmod(0o400)
|
|
files[trust] = path
|
|
base = {key: value for key, value in HEADERS.items() if key != "X-Hux-Relay-Key"}
|
|
for trust, surface in (("router", "chat"), ("relay", "telegram"), ("worker", "worker")):
|
|
headers = {**base, "X-Hux-Trust": trust, "X-Hux-Surface": surface, "X-Hux-Relay-Key": f"{trust}-secret"}
|
|
env = {f"HUX_{trust.upper()}_KEY_FILE": str(files[trust])}
|
|
assert identity.resolve(headers, env).trust == trust
|
|
wrong = files["router" if trust != "router" else "worker"]
|
|
with pytest.raises(errors.Unauthorized):
|
|
identity.resolve(headers, {f"HUX_{trust.upper()}_KEY_FILE": str(wrong)})
|
|
files["router"].chmod(0o440)
|
|
with pytest.raises(errors.Unauthorized):
|
|
identity.resolve({**base, "X-Hux-Relay-Key": "router-secret"}, {"HUX_ROUTER_KEY_FILE": str(files["router"])})
|
|
|
|
|
|
def _trusted_headers(trust: str, subject: str = "usr_0123456789abcdef") -> dict[str, str]:
|
|
"""Build headers for one separately keyed authenticated hop."""
|
|
surfaces = {"router": "chat", "relay": "telegram", "worker": "worker"}
|
|
return {
|
|
**HEADERS,
|
|
"X-Hux-Subject": subject,
|
|
"X-Hux-Trust": trust,
|
|
"X-Hux-Surface": surfaces[trust],
|
|
"X-Hux-Relay-Key": f"{trust}-key",
|
|
}
|
|
|
|
|
|
def _binding_env(path: Path) -> dict[str, str]:
|
|
"""Return the three-key environment used by subject-binding tests."""
|
|
return {
|
|
"HUX_ROUTER_KEY": "router-key",
|
|
"HUX_RELAY_KEY": "relay-key",
|
|
"HUX_WORKER_KEY": "worker-key",
|
|
"HUX_SUBJECT_BINDING_FILE": str(path),
|
|
}
|
|
|
|
|
|
def test_edge_hop_binds_subject_before_worker_can_assert_it(tmp_path):
|
|
"""Only a keyed router/relay may initialize the file; all later callers must match it."""
|
|
binding = tmp_path / "subject"
|
|
environ = _binding_env(binding)
|
|
with pytest.raises(errors.Unauthorized, match="not bound"):
|
|
identity.resolve(_trusted_headers("worker"), environ)
|
|
assert identity.resolve(_trusted_headers("router"), environ).subject == "usr_0123456789abcdef"
|
|
assert binding.read_text().strip() == "usr_0123456789abcdef"
|
|
assert binding.stat().st_mode & 0o777 == 0o440
|
|
assert identity.resolve(_trusted_headers("worker"), environ).trust == "worker"
|
|
other = "usr_fedcba9876543210"
|
|
for trust in ("router", "relay", "worker"):
|
|
with pytest.raises(errors.Unauthorized, match="does not match"):
|
|
identity.resolve(_trusted_headers(trust, other), environ)
|
|
|
|
|
|
def test_relay_can_initialize_binding_and_invalid_files_fail_closed(tmp_path):
|
|
"""Relay is an authoritative edge; malformed, linked, weak, and unavailable files never authenticate."""
|
|
relay_binding = tmp_path / "relay-subject"
|
|
assert identity.resolve(_trusted_headers("relay"), _binding_env(relay_binding)).trust == "relay"
|
|
for name, payload, mode in (
|
|
("empty", b"", 0o440),
|
|
("oversized", b"x" * (identity.MAX_SUBJECT_BYTES + 1), 0o440),
|
|
("unicode", b"\xff", 0o440),
|
|
("malformed", b"brad@example.test", 0o440),
|
|
("weak", b"usr_0123456789abcdef", 0o444),
|
|
):
|
|
binding = tmp_path / name
|
|
binding.write_bytes(payload)
|
|
binding.chmod(mode)
|
|
with pytest.raises(errors.Unauthorized):
|
|
identity.resolve(_trusted_headers("router"), _binding_env(binding))
|
|
link = tmp_path / "linked"
|
|
link.symlink_to(relay_binding)
|
|
with pytest.raises(errors.Unauthorized):
|
|
identity.resolve(_trusted_headers("router"), _binding_env(link))
|
|
with pytest.raises(errors.Unauthorized, match="unavailable"):
|
|
identity.resolve(_trusted_headers("router"), _binding_env(tmp_path / "missing" / "subject"))
|
|
|
|
|
|
def test_concurrent_edge_binding_has_exactly_one_subject(tmp_path):
|
|
"""First-writer publication is complete and immutable even when two edge hops race."""
|
|
binding = tmp_path / "subject"
|
|
environ = _binding_env(binding)
|
|
barrier = threading.Barrier(2)
|
|
outcomes: list[tuple[str, bool]] = []
|
|
|
|
def bind(trust: str, subject: str) -> None:
|
|
barrier.wait()
|
|
try:
|
|
identity.resolve(_trusted_headers(trust, subject), environ)
|
|
except errors.Unauthorized:
|
|
outcomes.append((subject, False))
|
|
else:
|
|
outcomes.append((subject, True))
|
|
|
|
attempts = (
|
|
threading.Thread(target=bind, args=("router", "usr_0123456789abcdef")),
|
|
threading.Thread(target=bind, args=("relay", "usr_fedcba9876543210")),
|
|
)
|
|
for attempt in attempts:
|
|
attempt.start()
|
|
for attempt in attempts:
|
|
attempt.join()
|
|
bound = binding.read_text().strip()
|
|
assert sorted(ok for _, ok in outcomes) == [False, True]
|
|
assert bound in {subject for subject, ok in outcomes if ok}
|
|
|
|
|
|
def test_capability_flags_are_bound_to_the_routes_really_registered():
|
|
"""A configured card stays off when a declared route is absent; undeclared routes fail startup."""
|
|
all_routes = {card: set(routes) for card, routes in flags.CARD_ROUTES.items() if routes}
|
|
bound = flags.Flags({"HUX_FLAGS": ALL_ON})
|
|
missing = {card: set(routes) for card, routes in all_routes.items()}
|
|
missing["HUX-01"].remove("/hux/v1/conversations/{id}/events/stream")
|
|
bound.bind_routes(missing)
|
|
assert not bound.enabled("HUX-01") and bound.enabled("HUX-11")
|
|
with pytest.raises(ValueError, match="unknown cards"):
|
|
bound.bind_routes({"HUX-99": {"/hux/v1/nope"}})
|
|
with pytest.raises(ValueError, match="undeclared routes"):
|
|
bound.bind_routes({"HUX-11": {"/hux/v1/nope"}})
|
|
|
|
|
|
def _raw_request(address: tuple[str, int], request: bytes) -> bytes:
|
|
"""Send one HTTP/1.0 request and return its complete close-delimited response."""
|
|
with socket.create_connection(address, timeout=1) as client:
|
|
client.sendall(request)
|
|
chunks = []
|
|
while chunk := client.recv(4096):
|
|
chunks.append(chunk)
|
|
return b"".join(chunks)
|
|
|
|
|
|
def test_declared_body_is_rejected_before_socket_read(tmp_path):
|
|
"""An oversized Content-Length receives 413 without waiting for the claimed body."""
|
|
server = _start(_router(tmp_path))
|
|
try:
|
|
conn = HTTPConnection("127.0.0.1", server.server_address[1], timeout=1)
|
|
conn.putrequest("POST", "/hux/v1/capabilities")
|
|
conn.putheader("Content-Length", str(1024 * 1024 + 1))
|
|
conn.endheaders()
|
|
reply = conn.getresponse()
|
|
body = json.loads(reply.read())
|
|
assert (reply.status, body["code"]) == (413, "too_large")
|
|
assert reply.getheader("Cache-Control") == "no-store"
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
|
|
|
|
def test_incomplete_body_times_out_and_transfer_encoding_is_rejected(tmp_path):
|
|
"""Accepted sockets have a deadline, and unsupported framing fails closed."""
|
|
server = _start(_router(tmp_path, HUX_REQUEST_TIMEOUT_SECONDS="0.1"))
|
|
try:
|
|
reply = _raw_request(
|
|
server.server_address,
|
|
b"POST /hux/v1/capabilities HTTP/1.0\r\nContent-Length: 2\r\n\r\n{",
|
|
)
|
|
assert b" 400 " in reply and b"request body is incomplete or timed out" in reply
|
|
conn = HTTPConnection("127.0.0.1", server.server_address[1], timeout=1)
|
|
conn.putrequest("POST", "/hux/v1/capabilities")
|
|
conn.putheader("Transfer-Encoding", "chunked")
|
|
conn.endheaders()
|
|
assert conn.getresponse().status == 400
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
|
|
|
|
def test_malformed_duplicate_and_absurd_content_lengths_fail_closed(tmp_path):
|
|
"""Ambiguous or computationally large length headers never reach ``int`` or a body allocation."""
|
|
server = _start(_router(tmp_path))
|
|
try:
|
|
malformed = _raw_request(
|
|
server.server_address,
|
|
b"POST /hux/v1/capabilities HTTP/1.0\r\nContent-Length: nope\r\n\r\n",
|
|
)
|
|
duplicate = _raw_request(
|
|
server.server_address,
|
|
b"POST /hux/v1/capabilities HTTP/1.0\r\nContent-Length: 0\r\nContent-Length: 0\r\n\r\n",
|
|
)
|
|
absurd = _raw_request(
|
|
server.server_address,
|
|
b"POST /hux/v1/capabilities HTTP/1.0\r\nContent-Length: 99999999999\r\n\r\n",
|
|
)
|
|
assert b" 400 " in malformed and b" 400 " in duplicate
|
|
assert b" 413 " in absurd
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
|
|
|
|
def test_rate_limit_is_per_subject_and_method_class(tmp_path):
|
|
"""Read and write buckets are distinct and return a bounded Retry-After."""
|
|
router = _router(tmp_path, HUX_READS_PER_MINUTE="1", HUX_WRITES_PER_MINUTE="1")
|
|
first = router.dispatch("GET", "/hux/v1/capabilities", HEADERS, b"")
|
|
limited = router.dispatch("GET", "/hux/v1/capabilities", HEADERS, b"")
|
|
write = router.dispatch("POST", "/hux/v1/capabilities", HEADERS, b"{}")
|
|
assert first.status == 200
|
|
assert (limited.status, limited.body["code"]) == (429, "rate_limited")
|
|
assert 1 <= int(limited.headers["Retry-After"]) <= 60
|
|
assert write.status == 405, "method routing precedes the write rate bucket"
|
|
live = _start(_router(tmp_path / "live", HUX_READS_PER_MINUTE="1"))
|
|
try:
|
|
conn = HTTPConnection("127.0.0.1", live.server_address[1], timeout=1)
|
|
conn.request("GET", "/hux/v1/capabilities", headers=HEADERS)
|
|
conn.getresponse().read()
|
|
conn.request("GET", "/hux/v1/capabilities", headers=HEADERS)
|
|
reply = conn.getresponse()
|
|
reply.read()
|
|
assert reply.status == 429 and reply.getheader("Retry-After")
|
|
assert reply.getheader("Cache-Control") == "no-store"
|
|
finally:
|
|
live.shutdown()
|
|
live.server_close()
|
|
|
|
|
|
def test_limiter_expires_windows_and_invalid_settings_use_safe_defaults(tmp_path, monkeypatch):
|
|
"""Expired subjects are evicted and malformed operator settings cannot disable bounds."""
|
|
moments = iter((0.0, 0.0, 61.0))
|
|
limiter = RateLimiter(1, 1, clock=lambda: next(moments))
|
|
assert limiter.check("a", "GET") is None
|
|
assert limiter.check("a", "GET") == 60
|
|
assert limiter.check("a", "GET") is None
|
|
monkeypatch.setattr(hux_http, "MAX_RATE_BUCKETS", 1)
|
|
bounded = RateLimiter(2, 2, clock=lambda: 0.0)
|
|
assert bounded.check("a", "GET") is None
|
|
assert bounded.check("b", "GET") == 60
|
|
router = _router(
|
|
tmp_path,
|
|
HUX_READS_PER_MINUTE="nan",
|
|
HUX_WRITES_PER_MINUTE="0",
|
|
HUX_REQUEST_TIMEOUT_SECONDS="forever",
|
|
)
|
|
assert router.rate_limiter.limits == {"read": 300, "write": 30}
|
|
assert router.request_timeout == DEFAULT_REQUEST_TIMEOUT_SECONDS
|
|
|
|
|
|
def test_health_and_json_are_no_store_and_health_refuses_bodies(tmp_path):
|
|
"""Sensitive JSON never enters browser caches; health cannot be used for body smuggling."""
|
|
server = _start(_router(tmp_path))
|
|
try:
|
|
conn = HTTPConnection("127.0.0.1", server.server_address[1], timeout=1)
|
|
conn.request("GET", "/healthz")
|
|
reply = conn.getresponse()
|
|
reply.read()
|
|
assert reply.getheader("Cache-Control") == "no-store"
|
|
assert reply.getheader("X-Content-Type-Options") == "nosniff"
|
|
conn.request("GET", "/healthz", body=b"x", headers={"Content-Length": "1"})
|
|
assert conn.getresponse().status == 413
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
|
|
|
|
def test_unknown_routes_are_limited_and_query_fields_are_bounded(tmp_path):
|
|
"""Authenticated misses consume their bucket and oversized query maps fail as typed 400s."""
|
|
router = _router(tmp_path, HUX_READS_PER_MINUTE="1")
|
|
missing = router.dispatch("GET", "/hux/v1/nope", HEADERS, b"")
|
|
limited = router.dispatch("GET", "/hux/v1/nope", HEADERS, b"")
|
|
assert missing.status == 404
|
|
assert (limited.status, limited.body["code"]) == (429, "rate_limited")
|
|
roomy = _router(tmp_path / "roomy")
|
|
query = "&".join(f"x{index}=1" for index in range(hux_http.MAX_QUERY_FIELDS + 1))
|
|
response = roomy.dispatch("GET", f"/hux/v1/capabilities?{query}", HEADERS, b"")
|
|
assert (response.status, response.body["code"]) == (400, "invalid")
|