513 lines
18 KiB
Python
513 lines
18 KiB
Python
"""Fail-closed loopback HTTP shell for V2R1 ingress processors."""
|
|
from __future__ import annotations
|
|
|
|
import ipaddress
|
|
import json
|
|
import socket
|
|
import threading
|
|
import time
|
|
from collections.abc import Callable, Mapping
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from typing import Any
|
|
from urllib.parse import urlsplit
|
|
|
|
|
|
MAX_REQUEST_BYTES = 1024 * 1024
|
|
MAX_RESPONSE_BYTES = 4096
|
|
REQUEST_IO_TIMEOUT_SECONDS = 0.5
|
|
REQUEST_NETWORK_DEADLINE_SECONDS = 1.0
|
|
HEX_DIGITS = frozenset("0123456789abcdefABCDEF")
|
|
HEADER_NAME_BYTES = frozenset(
|
|
b"!#$%&'*+-.^_`|~0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"
|
|
)
|
|
INTERNAL_ERROR_BODY = b'{"status":"INTERNAL_ERROR"}'
|
|
MUTATION_ROUTES = frozenset(
|
|
{"/v1/ingress", "/v1/receipts/status", "/v1/outbound-receipts"}
|
|
)
|
|
Processor = Callable[[str, bytes], tuple[int, Mapping[str, Any]]]
|
|
|
|
|
|
class RequestRejected(Exception):
|
|
def __init__(self, status: int, payload_status: str) -> None:
|
|
super().__init__(payload_status)
|
|
self.status = status
|
|
self.payload_status = payload_status
|
|
|
|
|
|
class IngressHTTPServer(ThreadingHTTPServer):
|
|
"""Threaded loopback server with an optional externally supplied processor."""
|
|
|
|
daemon_threads = False
|
|
block_on_close = True
|
|
|
|
def __init__(
|
|
self,
|
|
address: tuple[str, int],
|
|
handler: type[BaseHTTPRequestHandler],
|
|
processor: Processor | None,
|
|
) -> None:
|
|
self._closing = threading.Event()
|
|
self._lifecycle_lock = threading.RLock()
|
|
self._connections_lock = threading.Lock()
|
|
self._connections: set[socket.socket] = set()
|
|
self._server_close_lock = threading.Lock()
|
|
self.processor = processor
|
|
super().__init__(address, handler)
|
|
|
|
@property
|
|
def is_closing(self) -> bool:
|
|
return self._closing.is_set()
|
|
|
|
def get_request(self) -> tuple[socket.socket, tuple[str, int]]:
|
|
"""Register each accepted socket before handler scheduling can begin."""
|
|
connection, client_address = super().get_request()
|
|
with self._lifecycle_lock:
|
|
if self._closing.is_set():
|
|
reject = True
|
|
else:
|
|
with self._connections_lock:
|
|
self._connections.add(connection)
|
|
reject = False
|
|
if reject:
|
|
self._shutdown_and_close_connection(connection)
|
|
raise OSError("server is closing")
|
|
return connection, client_address
|
|
|
|
def unregister_connection(self, connection: socket.socket) -> None:
|
|
with self._connections_lock:
|
|
self._connections.discard(connection)
|
|
|
|
def _shutdown_and_close_connection(self, connection: socket.socket) -> None:
|
|
try:
|
|
connection.shutdown(socket.SHUT_RDWR)
|
|
except OSError:
|
|
pass
|
|
try:
|
|
connection.close()
|
|
except OSError:
|
|
pass
|
|
finally:
|
|
self.unregister_connection(connection)
|
|
|
|
def _close_accepted_connections(self) -> None:
|
|
with self._connections_lock:
|
|
connections = tuple(self._connections)
|
|
for connection in connections:
|
|
self._shutdown_and_close_connection(connection)
|
|
|
|
def process_request(
|
|
self,
|
|
request: socket.socket,
|
|
client_address: tuple[str, int],
|
|
) -> None:
|
|
"""Linearize request admission with the transition to closing."""
|
|
with self._lifecycle_lock:
|
|
if self._closing.is_set():
|
|
reject = True
|
|
else:
|
|
reject = False
|
|
try:
|
|
super().process_request(request, client_address)
|
|
except Exception:
|
|
self._shutdown_and_close_connection(request)
|
|
raise
|
|
if reject:
|
|
self._shutdown_and_close_connection(request)
|
|
|
|
def shutdown_request(self, request: socket.socket) -> None:
|
|
try:
|
|
request.shutdown(socket.SHUT_RDWR)
|
|
except OSError:
|
|
pass
|
|
try:
|
|
super().shutdown_request(request)
|
|
finally:
|
|
self.unregister_connection(request)
|
|
|
|
def call_processor(self, path: str, body: bytes) -> tuple[bool, Any]:
|
|
"""Start processing atomically with respect to server shutdown."""
|
|
with self._lifecycle_lock:
|
|
if self._closing.is_set() or self.processor is None:
|
|
return False, None
|
|
return True, self.processor(path, body)
|
|
|
|
def _begin_closing(self) -> None:
|
|
with self._lifecycle_lock:
|
|
self._closing.set()
|
|
|
|
def shutdown(self) -> None:
|
|
self._begin_closing()
|
|
self._close_accepted_connections()
|
|
super().shutdown()
|
|
|
|
def server_close(self) -> None:
|
|
"""Interrupt accepted sockets before the base class joins handlers."""
|
|
with self._server_close_lock:
|
|
self._begin_closing()
|
|
self._close_accepted_connections()
|
|
super().server_close()
|
|
|
|
def handle_error(
|
|
self,
|
|
request: socket.socket,
|
|
client_address: tuple[str, int],
|
|
) -> None:
|
|
"""Close unexpected handler failures without printing request data."""
|
|
try:
|
|
request.shutdown(socket.SHUT_RDWR)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
class _ValidatingHeaderReader:
|
|
"""Validate raw header lines before the standard library normalizes them."""
|
|
|
|
def __init__(self, handler: "IngressRequestHandler", reader: Any) -> None:
|
|
self._handler = handler
|
|
self._reader = reader
|
|
|
|
def readline(self, limit: int = -1) -> bytes:
|
|
try:
|
|
line = self._reader.readline(limit)
|
|
except TimeoutError as exc:
|
|
raise RequestRejected(408, "REQUEST_TIMEOUT") from exc
|
|
self._handler._ensure_network_read_active()
|
|
self._validate(line)
|
|
return line
|
|
|
|
@staticmethod
|
|
def _validate(line: bytes) -> None:
|
|
if line == b"\r\n":
|
|
return
|
|
if not line or not line.endswith(b"\r\n"):
|
|
raise RequestRejected(400, "BAD_REQUEST")
|
|
raw = line[:-2]
|
|
if not raw or raw[:1] in {b" ", b"\t"}:
|
|
raise RequestRejected(400, "BAD_REQUEST")
|
|
separator = raw.find(b":")
|
|
if separator <= 0:
|
|
raise RequestRejected(400, "BAD_REQUEST")
|
|
name = raw[:separator]
|
|
value = raw[separator + 1 :]
|
|
if any(byte not in HEADER_NAME_BYTES for byte in name):
|
|
raise RequestRejected(400, "BAD_REQUEST")
|
|
if any((byte < 0x20 and byte != 0x09) or byte == 0x7F for byte in value):
|
|
raise RequestRejected(400, "BAD_REQUEST")
|
|
if name.lower() == b"content-length":
|
|
numeric = value[1:] if value.startswith(b" ") else value
|
|
if not numeric or any(byte < 0x30 or byte > 0x39 for byte in numeric):
|
|
raise RequestRejected(400, "BAD_REQUEST")
|
|
|
|
|
|
class IngressRequestHandler(BaseHTTPRequestHandler):
|
|
"""Bounded JSON transport; application processing is injected later."""
|
|
|
|
server: IngressHTTPServer
|
|
|
|
def setup(self) -> None:
|
|
super().setup()
|
|
self.connection.settimeout(REQUEST_IO_TIMEOUT_SECONDS)
|
|
self._network_read_lock = threading.Lock()
|
|
self._network_read_expired = threading.Event()
|
|
self._network_read_deadline: float | None = None
|
|
self._network_read_timer: threading.Timer | None = None
|
|
self._network_read_active = False
|
|
|
|
def finish(self) -> None:
|
|
self._complete_network_read()
|
|
try:
|
|
super().finish()
|
|
except OSError:
|
|
pass
|
|
|
|
def _start_network_read(self) -> None:
|
|
self._complete_network_read()
|
|
timer = threading.Timer(
|
|
REQUEST_NETWORK_DEADLINE_SECONDS,
|
|
self._expire_network_read,
|
|
)
|
|
timer.name = f"aios-ingress-read-deadline-{id(self)}"
|
|
timer.daemon = True
|
|
with self._network_read_lock:
|
|
self._network_read_expired.clear()
|
|
self._network_read_deadline = (
|
|
time.monotonic() + REQUEST_NETWORK_DEADLINE_SECONDS
|
|
)
|
|
self._network_read_active = True
|
|
self._network_read_timer = timer
|
|
timer.start()
|
|
|
|
def _expire_network_read(self) -> None:
|
|
with self._network_read_lock:
|
|
if not self._network_read_active:
|
|
return
|
|
self._network_read_expired.set()
|
|
self.close_connection = True
|
|
try:
|
|
self.connection.shutdown(socket.SHUT_RD)
|
|
except OSError:
|
|
pass
|
|
|
|
def _complete_network_read(self) -> None:
|
|
timer: threading.Timer | None
|
|
with getattr(self, "_network_read_lock", threading.Lock()):
|
|
self._network_read_active = False
|
|
self._network_read_deadline = None
|
|
timer = self._network_read_timer
|
|
self._network_read_timer = None
|
|
if timer is not None:
|
|
timer.cancel()
|
|
if timer is not threading.current_thread():
|
|
timer.join(timeout=REQUEST_IO_TIMEOUT_SECONDS)
|
|
|
|
def _ensure_network_read_active(self) -> None:
|
|
if self.server.is_closing:
|
|
raise RequestRejected(503, "SERVICE_UNAVAILABLE")
|
|
deadline = self._network_read_deadline
|
|
if self._network_read_expired.is_set() or (
|
|
deadline is not None and time.monotonic() >= deadline
|
|
):
|
|
self._network_read_expired.set()
|
|
raise RequestRejected(408, "REQUEST_TIMEOUT")
|
|
|
|
def parse_request(self) -> bool:
|
|
original_reader = self.rfile
|
|
self.rfile = _ValidatingHeaderReader(self, original_reader) # type: ignore[assignment]
|
|
try:
|
|
return super().parse_request()
|
|
finally:
|
|
self.rfile = original_reader
|
|
|
|
def handle_one_request(self) -> None:
|
|
self._start_network_read()
|
|
try:
|
|
super().handle_one_request()
|
|
except RequestRejected as rejection:
|
|
self.close_connection = True
|
|
self._send_safely(rejection.status, {"status": rejection.payload_status})
|
|
except TimeoutError:
|
|
self.close_connection = True
|
|
self._send_safely(408, {"status": "REQUEST_TIMEOUT"})
|
|
except (BrokenPipeError, ConnectionResetError, OSError):
|
|
self.close_connection = True
|
|
except Exception:
|
|
self.close_connection = True
|
|
self._send_safely(500, {"status": "INTERNAL_ERROR"})
|
|
finally:
|
|
self._complete_network_read()
|
|
|
|
def send_error(
|
|
self,
|
|
code: int,
|
|
message: str | None = None,
|
|
explain: str | None = None,
|
|
) -> None:
|
|
"""Replace BaseHTTPRequestHandler HTML errors with bounded JSON."""
|
|
self.close_connection = True
|
|
if code == 501:
|
|
self._method_not_allowed()
|
|
return
|
|
self._send(int(code), {"status": "HTTP_ERROR"})
|
|
|
|
def log_message(self, format: str, *args: object) -> None:
|
|
return
|
|
|
|
def _send_safely(self, status: int, payload: Mapping[str, Any]) -> None:
|
|
try:
|
|
self._send(status, payload)
|
|
except OSError:
|
|
self.close_connection = True
|
|
|
|
def _path(self) -> str:
|
|
target = self.path
|
|
if not target or any(
|
|
ord(character) < 0x20 or ord(character) == 0x7F for character in target
|
|
):
|
|
raise RequestRejected(400, "BAD_REQUEST")
|
|
for index, character in enumerate(target):
|
|
if character == "%" and (
|
|
index + 2 >= len(target)
|
|
or any(value not in HEX_DIGITS for value in target[index + 1 : index + 3])
|
|
):
|
|
raise RequestRejected(400, "BAD_REQUEST")
|
|
try:
|
|
parts = urlsplit(target)
|
|
except ValueError as exc:
|
|
raise RequestRejected(400, "BAD_REQUEST") from exc
|
|
if parts.fragment:
|
|
raise RequestRejected(400, "BAD_REQUEST")
|
|
if target.startswith("/"):
|
|
if parts.scheme or parts.netloc:
|
|
raise RequestRejected(400, "BAD_REQUEST")
|
|
elif parts.scheme in {"http", "https"} and parts.netloc:
|
|
if parts.hostname is None:
|
|
raise RequestRejected(400, "BAD_REQUEST")
|
|
try:
|
|
_ = parts.port
|
|
except ValueError as exc:
|
|
raise RequestRejected(400, "BAD_REQUEST") from exc
|
|
else:
|
|
raise RequestRejected(400, "BAD_REQUEST")
|
|
return parts.path or "/"
|
|
|
|
@staticmethod
|
|
def _encoded_response(
|
|
status: object,
|
|
payload: object,
|
|
) -> tuple[int, bytes]:
|
|
if (
|
|
type(status) is not int
|
|
or not 200 <= status <= 599
|
|
or status in {204, 205, 304}
|
|
or not isinstance(payload, Mapping)
|
|
):
|
|
return 500, INTERNAL_ERROR_BODY
|
|
try:
|
|
body = json.dumps(
|
|
dict(payload),
|
|
ensure_ascii=False,
|
|
separators=(",", ":"),
|
|
sort_keys=True,
|
|
allow_nan=False,
|
|
).encode("utf-8")
|
|
except Exception:
|
|
return 500, INTERNAL_ERROR_BODY
|
|
if len(body) > MAX_RESPONSE_BYTES:
|
|
return 500, INTERNAL_ERROR_BODY
|
|
return status, body
|
|
|
|
def _send(self, status: int, payload: Mapping[str, Any]) -> None:
|
|
self._complete_network_read()
|
|
status, body = self._encoded_response(status, payload)
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.send_header("Cache-Control", "no-store")
|
|
self.end_headers()
|
|
if getattr(self, "command", None) != "HEAD":
|
|
self.wfile.write(body)
|
|
|
|
def _method_not_allowed(self) -> None:
|
|
self._send(405, {"status": "METHOD_NOT_ALLOWED"})
|
|
|
|
def _reject(self, status: int, payload_status: str) -> None:
|
|
self.close_connection = True
|
|
self._send(status, {"status": payload_status})
|
|
|
|
def _content_length(self) -> int:
|
|
if self.headers.get_all("Transfer-Encoding", []):
|
|
raise RequestRejected(400, "BAD_REQUEST")
|
|
values = self.headers.get_all("Content-Length", [])
|
|
if len(values) > 1:
|
|
raise RequestRejected(400, "BAD_REQUEST")
|
|
if not values:
|
|
return 0
|
|
raw_length = values[0]
|
|
if (
|
|
not raw_length
|
|
or not raw_length.isascii()
|
|
or not raw_length.isdigit()
|
|
):
|
|
raise RequestRejected(400, "BAD_REQUEST")
|
|
try:
|
|
length = int(raw_length)
|
|
except ValueError as exc:
|
|
raise RequestRejected(400, "BAD_REQUEST") from exc
|
|
if length > MAX_REQUEST_BYTES:
|
|
raise RequestRejected(413, "REQUEST_TOO_LARGE")
|
|
return length
|
|
|
|
def _read_exact_body(self, length: int) -> bytes:
|
|
try:
|
|
body = self.rfile.read(length)
|
|
except TimeoutError as exc:
|
|
raise RequestRejected(408, "REQUEST_TIMEOUT") from exc
|
|
self._ensure_network_read_active()
|
|
if len(body) != length:
|
|
raise RequestRejected(400, "BAD_REQUEST")
|
|
return body
|
|
|
|
def do_GET(self) -> None:
|
|
path = self._path()
|
|
if path == "/health/live":
|
|
self._send(200, {"status": "LIVE"})
|
|
elif path == "/health/ready":
|
|
if self.server.processor is None:
|
|
self._send(503, {"status": "SERVICE_NOT_CONFIGURED"})
|
|
else:
|
|
self._send(200, {"status": "READY"})
|
|
elif path in MUTATION_ROUTES:
|
|
self._method_not_allowed()
|
|
else:
|
|
self._send(404, {"status": "NOT_FOUND"})
|
|
|
|
def do_POST(self) -> None:
|
|
path = self._path()
|
|
if path not in MUTATION_ROUTES:
|
|
if path in {"/health/live", "/health/ready"}:
|
|
self._method_not_allowed()
|
|
else:
|
|
self._send(404, {"status": "NOT_FOUND"})
|
|
return
|
|
try:
|
|
length = self._content_length()
|
|
except RequestRejected as rejection:
|
|
self._reject(rejection.status, rejection.payload_status)
|
|
return
|
|
processor = self.server.processor
|
|
if processor is None:
|
|
self._send(503, {"status": "SERVICE_NOT_CONFIGURED"})
|
|
return
|
|
if self.server.is_closing:
|
|
self._reject(503, "SERVICE_UNAVAILABLE")
|
|
return
|
|
try:
|
|
body = self._read_exact_body(length)
|
|
except RequestRejected as rejection:
|
|
self._reject(rejection.status, rejection.payload_status)
|
|
return
|
|
self._complete_network_read()
|
|
try:
|
|
started, result = self.server.call_processor(path, body)
|
|
if not started:
|
|
self._reject(503, "SERVICE_UNAVAILABLE")
|
|
return
|
|
status, payload = result
|
|
except Exception:
|
|
self._send(500, {"status": "INTERNAL_ERROR"})
|
|
return
|
|
self._send(status, payload)
|
|
|
|
do_DELETE = _method_not_allowed
|
|
do_HEAD = _method_not_allowed
|
|
do_OPTIONS = _method_not_allowed
|
|
do_PATCH = _method_not_allowed
|
|
do_PUT = _method_not_allowed
|
|
|
|
|
|
def build_server(
|
|
host: str = "127.0.0.1",
|
|
port: int = 18792,
|
|
*,
|
|
processor: Processor | None = None,
|
|
) -> IngressHTTPServer:
|
|
"""Build a loopback-only HTTP server without starting it."""
|
|
try:
|
|
address = ipaddress.ip_address(host)
|
|
except ValueError as exc:
|
|
raise ValueError("host must be a loopback IP address") from exc
|
|
if address.version != 4 or not address.is_loopback:
|
|
raise ValueError("host must be an IPv4 loopback address")
|
|
return IngressHTTPServer((host, port), IngressRequestHandler, processor)
|
|
|
|
|
|
def serve(host: str = "127.0.0.1", port: int = 18792) -> None:
|
|
"""Run the unconfigured loopback shell until interrupted."""
|
|
server = build_server(host, port)
|
|
try:
|
|
server.serve_forever()
|
|
except KeyboardInterrupt:
|
|
return
|
|
finally:
|
|
server.server_close()
|