"""Offline HTTP/RPC regressions: authentication, noVNC, avatars and long histories. Docker is replaced with an isolated fixture; no real containers or model APIs are used. """ import base64 import hashlib import json import os from pathlib import Path import socket import sqlite3 import struct import subprocess import tempfile import threading import time from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import urllib.error import urllib.request import zlib from cli_flow import BINARY class Viewer(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" def do_GET(self): if self.headers.get("Upgrade", "").lower() == "websocket": key = self.headers["Sec-WebSocket-Key"] + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" self.send_response(101) self.send_header("Upgrade", "websocket") self.send_header("Connection", "Upgrade") self.send_header("Sec-WebSocket-Accept", base64.b64encode(hashlib.sha1(key.encode()).digest()).decode()) self.end_headers() self.wfile.write(b"\x82\x0cRFB 003.008\n") self.wfile.flush() self.close_connection = True time.sleep(.1) else: body = b"ISOLATED VNC FIXTURE" self.send_response(200) self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def log_message(self, *_): pass def chunk(kind, value): return struct.pack("!I", len(value)) + kind + value + struct.pack("!I", zlib.crc32(kind + value) & 0xffffffff) def main(): with tempfile.TemporaryDirectory(prefix="lazyboy-web-security-") as root, ThreadingHTTPServer(("127.0.0.1", 0), Viewer) as viewer: root = Path(root) threading.Thread(target=viewer.serve_forever, daemon=True).start() fakebin = root / "bin" fakebin.mkdir() docker = fakebin / "docker" docker.write_text('#!/bin/sh\ncase "$*" in\n*NetworkSettings.Ports*) printf "%s\\n" ' + str(viewer.server_port) + ' ;;\n"ps -a --format {{.Names}}") exit 0 ;;\n*) exit 1 ;;\nesac\n') docker.chmod(0o755) with socket.socket() as listener: listener.bind(("127.0.0.1", 0)) port = listener.getsockname()[1] data = root / "data" env = {**os.environ, "PATH": str(fakebin) + ":" + os.environ["PATH"], "LAZYBOY_DATA_DIR": str(data), "LAZYBOY_WEB_HOST": "127.0.0.1", "LAZYBOY_WEB_PORT": str(port), "LAZYBOY_WEB_TOKEN": "fixture-token", "LAZYBOY_API_KEY": "offline", "LAZYBOY_MODEL": "mock", "LAZYBOY_BASE_URL": "http://127.0.0.1:1", "LAZYBOY_TLS": "0"} origin = f"http://127.0.0.1:{port}" def http(path, method="GET", body=None, headers=None): request = urllib.request.Request(origin + path, data=body, method=method, headers=headers or {}) try: with urllib.request.urlopen(request, timeout=20) as response: return response.status, response.read(), response.headers except urllib.error.HTTPError as error: return error.code, error.read(), error.headers def websocket(cookie=None, source="https://untrusted.example", allowed=False): with socket.create_connection(("127.0.0.1", port), timeout=5) as stream: stream.settimeout(5) headers = ["GET /novnc/websockify HTTP/1.1", f"Host: 127.0.0.1:{port}", "Upgrade: websocket", "Connection: Upgrade", "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==", "Sec-WebSocket-Version: 13", f"Origin: {source}"] if cookie: headers.append("Cookie: " + cookie) stream.sendall(("\r\n".join(headers) + "\r\n\r\n").encode()) with stream.makefile("rb") as reader: status = reader.readline() assert (b" 101 " if allowed else b" 401 ") in status, status while reader.readline() != b"\r\n": pass if allowed: assert reader.read(14) == b"\x82\x0cRFB 003.008\n" with (root / "daemon.log").open("w") as log: process = subprocess.Popen([str(BINARY), "serve"], cwd=root, env=env, stdout=log, stderr=log) try: for _ in range(200): try: if http("/api/auth")[0] == 200: break except OSError: time.sleep(.05) else: raise AssertionError((root / "daemon.log").read_text()) auth = {"Authorization": "Bearer fixture-token", "Content-Type": "application/json"} assert json.loads(http("/api/auth")[1]) == {"required": True, "authenticated": False} for path in ("/api/health", "/novnc/vnc.html", "/novnc/core/rfb.js"): assert http(path)[0] == 401, path websocket() assert http("/api/auth", "POST", headers={"Authorization": "Bearer wrong"})[0] == 401 status, _, headers = http("/api/auth", "POST", headers=auth) assert status == 200 cookie_header = headers["Set-Cookie"] assert "HttpOnly" in cookie_header and "SameSite=Strict" in cookie_header and "Path=/" in cookie_header cookie = cookie_header.split(";", 1)[0] session = {"Cookie": cookie} assert http("/api/health", headers=session)[0] == 200 assert json.loads(http("/api/auth", headers=session)[1])["authenticated"] assert http("/api/health", headers={**session, "Origin": "https://untrusted.example"})[0] == 401 assert http("/novnc/vnc.html", headers=session)[0] == 200 websocket(cookie) websocket(cookie, origin, allowed=True) print("PASS API, noVNC HTTP and WebSocket require credentials; session origin is enforced", flush=True) status, body, _ = http("/api/agents", "POST", json.dumps({"name": "fixture", "description": "Offline fixture"}).encode(), auth) assert status == 200, body agent = json.loads(body)["id"] assert http(f"/api/agents/{agent}/events")[0] == 401 with urllib.request.urlopen(urllib.request.Request(origin + f"/api/agents/{agent}/events", headers=session), timeout=5) as stream: assert stream.status == 200 and stream.headers.get_content_type() == "text/event-stream" print("PASS browser session authenticates event streams", flush=True) width = 600 pixels = b"".join(b"\0" + os.urandom(width * 3) for _ in range(width)) png = b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", struct.pack("!2I5B", width, width, 8, 2, 0, 0, 0)) + chunk(b"IDAT", zlib.compress(pixels)) + chunk(b"IEND", b"") padded = png[:-12] + chunk(b"npAD", bytes(5 * 1024 * 1024 - len(png) - 12)) + png[-12:] for image in (png, padded): status, body, _ = http(f"/api/agents/{agent}/avatar", "PUT", image, {**session, "Content-Type": "image/png"}) assert status == 200, (len(image), status, body) status, body, _ = http(f"/api/agents/{agent}/avatar", headers=session) assert status == 200 and body == image assert len(padded) == 5 * 1024 * 1024 assert http(f"/api/agents/{agent}/avatar", "PUT", padded + b"x", {**session, "Content-Type": "image/png"})[0] == 413 print("PASS valid PNG uploads and downloads through 5 MiB; larger uploads return 413", flush=True) with sqlite3.connect(data / "team.sqlite3") as db: record = json.loads(db.execute("SELECT data FROM agents WHERE id=?", (agent,)).fetchone()[0]) record["conversation"] = [{"role": "system", "content": "fixture"}] + [{"role": "user", "content": f"{i}:" + "x" * 60000, "created_at": "2026-10-05T00:00:00Z"} for i in range(71)] db.execute("UPDATE agents SET data=? WHERE id=?", (json.dumps(record), agent)) status, body, _ = http(f"/api/agents/{agent}", headers=session) assert status == 200 and len(body) > 4 * 1024 * 1024, (status, body[:200]) transcript = json.loads(body)["transcript"] assert len(transcript) == 71 and transcript[-1]["content"] == "70:" + "x" * 60000 print("PASS history larger than 4 MiB arrives complete through daemon RPC", flush=True) finally: process.terminate() try: process.wait(timeout=10) except subprocess.TimeoutExpired: process.kill() process.wait() viewer.shutdown() if __name__ == "__main__": main()