scripts/qa-sandbox-http-ws.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""Bounded public-payload fixture. It has no file, environment or account routes."""
import argparse
import base64
import binascii
import hashlib
import json
import os
from pathlib import Path
import re
import resource
import socket
import socketserver
import struct
import threading
import time
from urllib.parse import urlsplit
PAYLOAD_LIMIT = 4096
HEADER_LIMIT = 4096
FRAME_LIMIT = 64
MESSAGE_LIMIT = 16
HEALTH = b'{"fixture":"firecrab-414-http-ws","ok":true}\n'
GUID = b"258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
HEADER_NAME = re.compile(r"[!#$%&'*+.^_`|~0-9A-Za-z-]+\Z")
CLOSE_CODES = {1000, 1001, 1002, 1003, 1007, 1008, 1009, 1010, 1011, 1012, 1013, 1014}
class ProtocolError(Exception):
"""A fixed HTTP status or WebSocket close code, never an input echo."""
class Reader:
"""Exact reads share one absolute deadline, including slow header/frame senders."""
def __init__(self, connection, seconds):
self.connection = connection
self.deadline = time.monotonic() + seconds
def read(self, length):
result = bytearray()
while len(result) < length:
remaining = self.deadline - time.monotonic()
if remaining <= 0:
raise TimeoutError
self.connection.settimeout(min(remaining, 2))
part = self.connection.recv(length - len(result))
if not part:
raise ConnectionError
result.extend(part)
return bytes(result)
def headers(self):
block = bytearray()
while len(block) < HEADER_LIMIT:
block.extend(self.read(1))
if block.endswith(b"\r\n\r\n"):
return bytes(block)
raise ProtocolError(431)
def request_headers(block):
"""Reject duplicate, folded and ambiguous HTTP framing before any upgrade."""
try:
lines = block.decode("ascii").split("\r\n")
method, path, version = lines[0].split(" ")
except (UnicodeError, ValueError):
raise ProtocolError(400) from None
if version != "HTTP/1.1":
raise ProtocolError(505)
headers = {}
for line in lines[1:-2]:
name, separator, value = line.partition(":")
if not separator or not HEADER_NAME.fullmatch(name):
raise ProtocolError(400)
value = value.strip(" \t")
name = name.lower()
if name in headers or any(ord(c) < 32 or ord(c) == 127 for c in value):
raise ProtocolError(400)
headers[name] = value
if not headers.get("host") or any(name in headers for name in
("transfer-encoding", "expect", "content-encoding")):
raise ProtocolError(400)
raw_length = headers.get("content-length", "0")
if not re.fullmatch(r"[0-9]+", raw_length):
raise ProtocolError(400)
length = int(raw_length)
if length > PAYLOAD_LIMIT:
raise ProtocolError(413)
return method, path, headers, length
def websocket_key(method, headers, length, allowed_origin):
"""Negotiate version 13 without extensions, compression or a subprotocol."""
if (method != "GET" or length or headers.get("upgrade", "").lower() != "websocket"
or "upgrade" not in {part.strip().lower() for part in
headers.get("connection", "").split(",")}
or headers.get("sec-websocket-version") != "13"):
raise ProtocolError(400)
if "origin" in headers and headers["origin"] != allowed_origin:
raise ProtocolError(403)
key = headers.get("sec-websocket-key", "")
try:
decoded = base64.b64decode(key, validate=True)
except (ValueError, binascii.Error):
raise ProtocolError(400) from None
if len(decoded) != 16 or base64.b64encode(decoded).decode("ascii") != key:
raise ProtocolError(400)
return base64.b64encode(hashlib.sha1(key.encode("ascii") + GUID).digest())
def frame(connection, opcode, payload):
"""The server always sends final, unmasked frames with a bounded payload."""
header = bytes([0x80 | opcode])
header += bytes([len(payload)]) if len(payload) < 126 else b"\x7e" + struct.pack("!H", len(payload))
connection.sendall(header + payload)
def read_frame(reader):
"""Client frames must be masked, minimally encoded and use only known opcodes."""
first, second = reader.read(2)
final, opcode = bool(first & 0x80), first & 15
if first & 0x70 or not second & 0x80 or opcode not in (0, 1, 2, 8, 9, 10):
raise ProtocolError(1002)
length = second & 127
if length == 126:
length = struct.unpack("!H", reader.read(2))[0]
if length < 126:
raise ProtocolError(1002)
elif length == 127:
length = struct.unpack("!Q", reader.read(8))[0]
if length < 65536 or length >> 63:
raise ProtocolError(1002)
if opcode >= 8 and (not final or length > 125):
raise ProtocolError(1002)
if length > PAYLOAD_LIMIT:
raise ProtocolError(1009)
mask = reader.read(4)
payload = reader.read(length)
return final, opcode, bytes(value ^ mask[index % 4] for index, value in enumerate(payload))
def validate_close(payload):
"""Reserved codes and invalid UTF-8 close reasons fail closed."""
if not payload:
return
if len(payload) == 1:
raise ProtocolError(1002)
code = struct.unpack("!H", payload[:2])[0]
if code not in CLOSE_CODES and not 3000 <= code <= 4999:
raise ProtocolError(1002)
payload[2:].decode("utf-8")
def websocket(connection, reader):
"""Echo complete text/binary messages; control frames can interrupt fragmentation."""
fragments, fragment_opcode, messages = bytearray(), None, 0
try:
for _ in range(FRAME_LIMIT):
final, opcode, payload = read_frame(reader)
if opcode == 8:
validate_close(payload)
frame(connection, 8, payload)
return
if opcode == 9:
frame(connection, 10, payload)
continue
if opcode == 10:
continue
if opcode == 0:
if fragment_opcode is None:
raise ProtocolError(1002)
fragments.extend(payload)
else:
if fragment_opcode is not None:
raise ProtocolError(1002)
fragment_opcode = opcode
fragments.extend(payload)
if len(fragments) > PAYLOAD_LIMIT:
raise ProtocolError(1009)
if not final:
continue
if fragment_opcode == 1:
fragments.decode("utf-8")
frame(connection, fragment_opcode, bytes(fragments))
fragments.clear()
fragment_opcode = None
messages += 1
if messages == MESSAGE_LIMIT:
break
frame(connection, 8, struct.pack("!H", 1008))
except UnicodeError:
frame(connection, 8, struct.pack("!H", 1007))
except ProtocolError as error:
frame(connection, 8, struct.pack("!H", error.args[0]))
class Handler(socketserver.BaseRequestHandler):
"""One HTTP request or one bounded WebSocket session; never serve a directory."""
def respond(self, status, body=b"", content_type="application/octet-stream", extra=b""):
headers = (f"HTTP/1.1 {status} QA\r\nContent-Length: {len(body)}\r\n"
f"Content-Type: {content_type}\r\nCache-Control: no-store\r\n"
"X-Content-Type-Options: nosniff\r\nConnection: close\r\n").encode("ascii")
self.request.sendall(headers + extra + b"\r\n" + body)
def handle(self):
reader = Reader(self.request, self.server.connection_seconds)
try:
try:
method, path, headers, length = request_headers(reader.headers())
if path == "/ws":
accept = websocket_key(method, headers, length, self.server.allowed_origin)
self.request.sendall(b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\n"
b"Connection: Upgrade\r\nCache-Control: no-store\r\n"
b"X-Content-Type-Options: nosniff\r\nSec-WebSocket-Accept: "
+ accept + b"\r\n\r\n")
websocket(self.request, reader)
return
if "upgrade" in headers or length and method != "POST":
raise ProtocolError(400)
if method == "GET" and path == "/health":
self.respond(200, HEALTH, "application/json")
elif method == "POST" and path == "/echo":
if "content-length" not in headers:
raise ProtocolError(411)
self.respond(200, reader.read(length))
else:
self.respond(404)
except ProtocolError as error:
self.respond(error.args[0])
except (OSError, ConnectionError, TimeoutError):
pass
class EchoServer(socketserver.ThreadingMixIn, socketserver.TCPServer):
"""Reject excess connections instead of creating an unbounded worker queue."""
allow_reuse_address = True
daemon_threads = True
request_queue_size = 4
def __init__(self, address, *, allowed_origin=None, connection_seconds=10):
self.allowed_origin = allowed_origin
self.connection_seconds = connection_seconds
self.slots = threading.BoundedSemaphore(4)
super().__init__(address, Handler)
def process_request(self, request, client_address):
if not self.slots.acquire(blocking=False):
self.shutdown_request(request)
return
try:
super().process_request(request, client_address)
except Exception:
self.slots.release()
raise
def process_request_thread(self, request, client_address):
try:
super().process_request_thread(request, client_address)
finally:
self.slots.release()
def handle_error(self, request, client_address):
# No request path, headers, payload, exception or environment enters logs.
pass
def confined_identity():
"""Read only this process's privilege metadata, never guest account or key files."""
status = dict(line.split(":", 1) for line in Path("/proc/self/status").read_text().splitlines()
if ":" in line)
if (os.getuid() == 0 or os.geteuid() == 0 or int(status["NoNewPrivs"].strip()) != 1
or int(status["CapEff"].strip(), 16)):
raise ValueError("Run as a non-root UID with NoNewPrivs=1 and CapEff=0")
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--host", choices=("127.0.0.1", "0.0.0.0"), default="127.0.0.1")
parser.add_argument("--port", type=int, default=8080)
parser.add_argument("--origin", help="Exact browser Origin; absent Origin is allowed for native QA clients")
parser.add_argument("--lifetime-seconds", type=int, default=300)
args = parser.parse_args()
if args.port != 0 and not 1024 <= args.port <= 65535 or not 1 <= args.lifetime_seconds <= 900:
parser.error("Use port 0 or 1024..65535 and a lifetime of 1..900 seconds")
if args.origin:
try:
origin = urlsplit(args.origin)
except ValueError:
parser.error("Origin must be a literal http(s) origin")
if (origin.scheme not in ("http", "https") or not origin.hostname or origin.username
or origin.password or origin.path or origin.query or origin.fragment
or any(c.isspace() or ord(c) < 32 for c in args.origin)):
parser.error("Origin must be a literal http(s) origin without credentials, path or query")
try:
confined_identity()
except (OSError, ValueError, KeyError):
parser.error("Run on Linux as a non-root UID with NoNewPrivs=1 and CapEff=0")
for kind, maximum in ((resource.RLIMIT_NOFILE, 64), (resource.RLIMIT_AS, 128 * 1024 * 1024),
(resource.RLIMIT_CPU, 15), (resource.RLIMIT_FSIZE, 0), (resource.RLIMIT_CORE, 0)):
resource.setrlimit(kind, (maximum, maximum))
with EchoServer((args.host, args.port), allowed_origin=args.origin) as server:
server.timeout = 0.2
print(json.dumps({"fixture": "firecrab-414-http-ws", "host": args.host,
"port": server.server_address[1], "uid": os.getuid(), "noNewPrivs": True}), flush=True)
deadline = time.monotonic() + args.lifetime_seconds
while time.monotonic() < deadline:
server.handle_request()
if __name__ == "__main__":
main()