diff --git a/docs/designs/local-web-console.md b/docs/designs/local-web-console.md index 1b2757a..2c1717f 100644 --- a/docs/designs/local-web-console.md +++ b/docs/designs/local-web-console.md @@ -190,6 +190,19 @@ index、偏好更新、Git 锁写入或其他 hydration mutation。 - C01 不发布 action receipt、ledger、gate 输出、agent argv、adapter 环境、仓库路径或 remote。 后续阶段若要扩充字段，必须先扩充显式 DTO 与脱敏测试，而不能把 Core dataclass 直接 JSON 化。 +### 4.2 C02 已落地的 listener 契约 + +C02 只提供无项目数据的静态 shell、一次性 session 交换和认证后的 `/api/v1/meta`： + +- `create_console_http_server()` 没有 host 参数，只能绑定 `127.0.0.1`；它对 request line、 + headers、body、并发数、读取时间和单请求总时限设置固定上限； +- 每个请求必须使用精确的 `Host: 127.0.0.1:`；转发 header、`Transfer-Encoding`、 + 非 origin-form target 与不允许的方法都 fail closed，且不会启用 CORS 或访问日志； +- bootstrap 仅在精确 same-origin JSON POST 中使用一次，随后换发独立的内存 bearer。session 没有 + cookie，使用 30 分钟 idle 与 8 小时 absolute 上限；退出 server 时会清空全部 session； +- C02 没有 CLI `console` 命令、浏览器打开、资源文件读取或 workspace API。它的唯一目的，是在 + 接入真实 read model 前先固定并测试本地 HTTP 安全边界。 + ## 5. 模块设计 建议模块边界： diff --git a/src/dyro/console/__init__.py b/src/dyro/console/__init__.py index c844e34..338c80f 100644 --- a/src/dyro/console/__init__.py +++ b/src/dyro/console/__init__.py @@ -5,5 +5,6 @@ """ from .read_model import workspace_envelope +from .server import create_console_http_server -__all__ = ["workspace_envelope"] +__all__ = ["create_console_http_server", "workspace_envelope"] diff --git a/src/dyro/console/server.py b/src/dyro/console/server.py new file mode 100644 index 0000000..c545d62 --- /dev/null +++ b/src/dyro/console/server.py @@ -0,0 +1,440 @@ +"""Loopback-only HTTP boundary for the read-only local Console.""" + +from __future__ import annotations + +from email.message import Message +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import json +import re +import socket +import threading +from typing import Any +from urllib.parse import urlsplit + +from .. import __version__ +from .session import ConsoleSessionStore, SessionRejected + + +HOST = "127.0.0.1" +REQUEST_LINE_LIMIT = 4 * 1024 +HEADER_LIMIT = 16 * 1024 +HEADER_LINE_LIMIT = 4 * 1024 +SESSION_BODY_LIMIT = 512 +READ_TIMEOUT_SECONDS = 5.0 +REQUEST_DEADLINE_SECONDS = 10.0 +MAX_CONCURRENT_REQUESTS = 8 +_CSP = ( + "default-src 'none'; script-src 'self'; style-src 'self'; img-src 'self'; " + "connect-src 'self'; worker-src 'none'; object-src 'none'; base-uri 'none'; " + "form-action 'none'; frame-ancestors 'none'" +) +_STATIC_SHELL = b""" +Dyro Console +

Dyro Console

Secure local session is starting.

""" +_TOKEN = re.compile(r"^[!#$%&'*+.^_`|~0-9A-Za-z-]+$") + + +def _json_bytes(value: object) -> bytes: + return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode( + "utf-8" + ) + + +class ConsoleHTTPServer(ThreadingHTTPServer): + """A bounded IPv4 loopback server with no request logging.""" + + address_family = socket.AF_INET + daemon_threads = True + request_queue_size = MAX_CONCURRENT_REQUESTS + + def __init__( + self, + *, + port: int, + bootstrap_secret: str | None, + session_store: ConsoleSessionStore | None, + max_concurrent_requests: int, + ) -> None: + self._request_slots = threading.BoundedSemaphore(max_concurrent_requests) + self.sessions = session_store + super().__init__((HOST, port), ConsoleRequestHandler) + if self.sessions is None: + self.sessions = ConsoleSessionStore(bootstrap_secret=bootstrap_secret) + self.origin = f"http://{HOST}:{self.server_port}" + + def get_request(self) -> tuple[Any, tuple[str, int]]: + request, client_address = super().get_request() + request.settimeout(READ_TIMEOUT_SECONDS) + return request, client_address + + def process_request(self, request: Any, client_address: tuple[str, int]) -> None: + if not self._request_slots.acquire(blocking=False): + request.close() + return + super().process_request(request, client_address) + + def process_request_thread(self, request: Any, client_address: tuple[str, int]) -> None: + timer = threading.Timer(REQUEST_DEADLINE_SECONDS, self._close_request, args=(request,)) + timer.daemon = True + timer.start() + try: + super().process_request_thread(request, client_address) + finally: + timer.cancel() + self._request_slots.release() + + @staticmethod + def _close_request(request: Any) -> None: + try: + request.shutdown(socket.SHUT_RDWR) + except OSError: + pass + try: + request.close() + except OSError: + pass + + def server_close(self) -> None: + if self.sessions is not None: + self.sessions.clear() + super().server_close() + + def handle_error(self, request: Any, client_address: tuple[str, int]) -> None: + """Never write a traceback, request text, or local paths to stderr.""" + del request, client_address + return + + +class ConsoleRequestHandler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + server_version = "DyroConsole" + sys_version = "" + + @property + def console(self) -> ConsoleHTTPServer: + return self.server # type: ignore[return-value] + + def log_message(self, format: str, *args: object) -> None: + return + + def send_error( + self, + code: int, + message: str | None = None, + explain: str | None = None, + ) -> None: + """Do not let stdlib parser messages reflect raw request material.""" + del message, explain + self._error(400 if 400 <= code < 500 else 500, "BAD_REQUEST") + + def handle_one_request(self) -> None: + """Apply the Console request-line ceiling before stdlib parsing. + + ``BaseHTTPRequestHandler`` otherwise accepts a much larger line before + application code can enforce the documented 4 KiB protocol boundary. + """ + try: + self.raw_requestline = self.rfile.readline(REQUEST_LINE_LIMIT + 1) + if len(self.raw_requestline) > REQUEST_LINE_LIMIT: + self.requestline = "" + self.request_version = "HTTP/1.1" + self.command = "" + self.close_connection = True + self._error(400, "BAD_REQUEST") + return + if not self.raw_requestline: + self.close_connection = True + return + if not self._parse_console_request(): + return + method = getattr(self, f"do_{self.command}", None) + if method is None: + self._reject_method() + else: + method() + self.wfile.flush() + except OSError: + self.close_connection = True + + def _parse_console_request(self) -> bool: + """Parse only strict HTTP/1.1 origin-form requests within fixed bounds.""" + if not self.raw_requestline.endswith(b"\r\n"): + self._bad_parse_request() + return False + try: + requestline = self.raw_requestline[:-2].decode("ascii") + except UnicodeDecodeError: + self._bad_parse_request() + return False + fields = requestline.split(" ") + if len(fields) != 3 or not all(fields) or not _TOKEN.fullmatch(fields[0]): + self._bad_parse_request() + return False + method, target, version = fields + if version != "HTTP/1.1" or not target.startswith("/") or target.startswith("//"): + self._bad_parse_request() + return False + self.requestline = requestline + self.command = method + self.path = target + self.request_version = version + headers = Message() + total = 0 + while True: + line = self.rfile.readline(HEADER_LINE_LIMIT + 1) + if not line or len(line) > HEADER_LINE_LIMIT: + self._bad_parse_request() + return False + total += len(line) + if total > HEADER_LIMIT: + self._bad_parse_request() + return False + if line == b"\r\n": + self.headers = headers + return True + if not line.endswith(b"\r\n") or line[:1] in {b" ", b"\t"}: + self._bad_parse_request() + return False + raw = line[:-2] + if any(byte < 0x20 or byte == 0x7F for byte in raw): + self._bad_parse_request() + return False + name, separator, value = raw.partition(b":") + if not separator or not name: + self._bad_parse_request() + return False + try: + decoded_name = name.decode("ascii") + decoded_value = value.decode("latin-1").strip() + except UnicodeDecodeError: + self._bad_parse_request() + return False + if not _TOKEN.fullmatch(decoded_name): + self._bad_parse_request() + return False + headers.add_header(decoded_name, decoded_value) + + def _bad_parse_request(self) -> None: + self.requestline = "" + self.request_version = "HTTP/1.1" + self.command = "" + self.close_connection = True + self._error(400, "BAD_REQUEST") + + def do_GET(self) -> None: + self._dispatch() + + def do_POST(self) -> None: + self._dispatch() + + def do_OPTIONS(self) -> None: + self._reject_method() + + def do_PUT(self) -> None: + self._reject_method() + + def do_PATCH(self) -> None: + self._reject_method() + + def do_DELETE(self) -> None: + self._reject_method() + + def do_HEAD(self) -> None: + self._reject_method() + + def do_TRACE(self) -> None: + self._reject_method() + + def do_CONNECT(self) -> None: + self._reject_method() + + def _dispatch(self) -> None: + if not self._validate_request_envelope(): + return + parsed = urlsplit(self.path) + if parsed.query or parsed.fragment or parsed.path != self.path: + self._error(400, "BAD_REQUEST") + return + if self.command == "GET" and parsed.path == "/": + if self._has_body(): + self._error(400, "BAD_REQUEST") + return + self._send(200, _STATIC_SHELL, "text/html; charset=utf-8") + return + if parsed.path == "/api/v1/session": + if self.command != "POST": + self._method_not_allowed() + return + self._exchange_session() + return + if parsed.path == "/api/v1/meta": + if self.command != "GET": + self._method_not_allowed() + return + if self._has_body(): + self._error(400, "BAD_REQUEST") + return + session = self._authorized_session() + if session is None: + return + self._json( + 200, + { + "schema_version": 1, + "data": { + "version": __version__, + "capabilities": [], + "session_expires_at": session.expires_at.isoformat(), + }, + }, + ) + return + if parsed.path.startswith("/api/"): + self._error(401, "UNAUTHORIZED") + return + self._error(404, "NOT_FOUND") + + def _validate_request_envelope(self) -> bool: + if len(self.raw_requestline) > REQUEST_LINE_LIMIT: + self._error(400, "BAD_REQUEST") + return False + if not self.path.startswith("/") or self.path.startswith("//"): + self._error(400, "BAD_REQUEST") + return False + try: + header_items = list(self.headers.items()) + except (TypeError, ValueError): + self._error(400, "BAD_REQUEST") + return False + if sum(len(name) + len(value) + 4 for name, value in header_items) > HEADER_LIMIT: + self._error(400, "BAD_REQUEST") + return False + hosts = self.headers.get_all("Host") or [] + if len(hosts) != 1 or hosts[0] != f"{HOST}:{self.console.server_port}": + self._error(400, "BAD_REQUEST") + return False + if self.headers.get_all("Transfer-Encoding"): + self._error(400, "BAD_REQUEST") + return False + if any(name.lower() == "forwarded" or name.lower().startswith("x-forwarded-") for name, _ in header_items): + self._error(400, "BAD_REQUEST") + return False + return True + + def _has_body(self) -> bool: + return bool(self.headers.get_all("Content-Length")) + + def _content_length(self) -> int | None: + values = self.headers.get_all("Content-Length") or [] + if len(values) != 1 or not values[0].isdigit(): + return None + value = int(values[0]) + return value if value <= SESSION_BODY_LIMIT else None + + def _valid_origin(self, *, required: bool) -> bool: + origins = self.headers.get_all("Origin") or [] + if required and len(origins) != 1: + return False + if origins and (len(origins) != 1 or origins[0] != self.console.origin): + return False + sites = self.headers.get_all("Sec-Fetch-Site") or [] + return not sites or (len(sites) == 1 and sites[0] == "same-origin") + + def _exchange_session(self) -> None: + if not self._valid_origin(required=True): + self._error(403, "ORIGIN_REJECTED") + return + content_types = self.headers.get_all("Content-Type") or [] + length = self._content_length() + if len(content_types) != 1 or content_types[0] != "application/json" or length is None: + self._error(400, "BAD_REQUEST") + return + try: + raw = self.rfile.read(length) + decoded = json.loads(raw.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError, OSError): + self._error(400, "BAD_REQUEST") + return + if not isinstance(decoded, dict) or set(decoded) != {"bootstrap"}: + self._error(400, "BAD_REQUEST") + return + try: + session = self.console.sessions.exchange(decoded["bootstrap"]) + except SessionRejected: + self._error(401, "UNAUTHORIZED") + return + self._json( + 201, + { + "schema_version": 1, + "bearer": session.token, + "session_expires_at": session.expires_at.isoformat(), + }, + ) + + def _authorized_session(self) -> object | None: + if not self._valid_origin(required=False): + self._error(403, "ORIGIN_REJECTED") + return None + values = self.headers.get_all("Authorization") or [] + if len(values) != 1 or not values[0].startswith("Bearer "): + self._error(401, "UNAUTHORIZED") + return None + try: + return self.console.sessions.authorize(values[0][len("Bearer ") :]) + except SessionRejected: + self._error(401, "UNAUTHORIZED") + return None + + def _method_not_allowed(self) -> None: + self._error(405, "METHOD_NOT_ALLOWED") + + def _reject_method(self) -> None: + if self._validate_request_envelope(): + self._method_not_allowed() + + def _json(self, status: int, payload: object) -> None: + self._send(status, _json_bytes(payload), "application/json; charset=utf-8") + + def _error(self, status: int, code: str) -> None: + self._json(status, {"schema_version": 1, "error": {"code": code}}) + + def _send(self, status: int, body: bytes, content_type: str) -> None: + # Every response terminates this HTTP/1.1 connection. In particular, + # rejected GET or unknown API requests must never leave an unread body + # that a later handler iteration could interpret as a request line. + self.close_connection = True + self.send_response(status) + self.send_header("Content-Type", content_type) + self.send_header("Content-Length", str(len(body))) + self.send_header("Cache-Control", "no-store") + self.send_header("Connection", "close") + self.send_header("X-Content-Type-Options", "nosniff") + self.send_header("Referrer-Policy", "no-referrer") + self.send_header("X-Frame-Options", "DENY") + self.send_header("Cross-Origin-Opener-Policy", "same-origin") + self.send_header("Content-Security-Policy", _CSP) + self.end_headers() + self.wfile.write(body) + + +def create_console_http_server( + *, + port: int = 0, + bootstrap_secret: str | None = None, + session_store: ConsoleSessionStore | None = None, + max_concurrent_requests: int = MAX_CONCURRENT_REQUESTS, +) -> ConsoleHTTPServer: + """Bind a Console server to IPv4 loopback only; no host override exists.""" + if type(port) is not int or not 0 <= port <= 65535: + raise ValueError("Console port 必须是 0 到 65535 之间的整数") + if type(max_concurrent_requests) is not int or max_concurrent_requests < 1: + raise ValueError("Console max_concurrent_requests 必须是正整数") + if session_store is not None and bootstrap_secret is not None: + raise ValueError("Console session_store 与 bootstrap_secret 不能同时指定") + return ConsoleHTTPServer( + port=port, + bootstrap_secret=bootstrap_secret, + session_store=session_store, + max_concurrent_requests=max_concurrent_requests, + ) diff --git a/src/dyro/console/session.py b/src/dyro/console/session.py new file mode 100644 index 0000000..660bebb --- /dev/null +++ b/src/dyro/console/session.py @@ -0,0 +1,128 @@ +"""One-time bootstrap exchange and in-memory Console sessions.""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +import hmac +import secrets +import threading +import time +from typing import Callable + + +BOOTSTRAP_TTL_SECONDS = 60.0 +BOOTSTRAP_MAX_FAILURES = 5 +SESSION_IDLE_TTL_SECONDS = 30 * 60.0 +SESSION_ABSOLUTE_TTL_SECONDS = 8 * 60 * 60.0 + + +class SessionRejected(Exception): + """A deliberately detail-free rejected bootstrap or bearer exchange.""" + + +@dataclass(frozen=True) +class SessionView: + token: str + expires_at: datetime + + +@dataclass +class _Session: + token: str + idle_deadline: float + absolute_deadline: float + absolute_expires_at: datetime + + +def _utc_now() -> datetime: + return datetime.now(timezone.utc) + + +class ConsoleSessionStore: + """Memory-only bearer state; no cookie, disk, or workspace side effect.""" + + def __init__( + self, + *, + bootstrap_secret: str | None = None, + monotonic_clock: Callable[[], float] = time.monotonic, + wall_clock: Callable[[], datetime] = _utc_now, + ) -> None: + secret = bootstrap_secret if bootstrap_secret is not None else secrets.token_urlsafe(32) + if not isinstance(secret, str) or len(secret) < 43: + raise ValueError("bootstrap secret 必须至少包含 256 bit 的 URL-safe 熵") + self._bootstrap_secret: str | None = secret + self._clock = monotonic_clock + self._wall_clock = wall_clock + self._bootstrap_deadline = monotonic_clock() + BOOTSTRAP_TTL_SECONDS + self._bootstrap_failures = 0 + self._sessions: dict[str, _Session] = {} + self._lock = threading.Lock() + + @property + def bootstrap_secret(self) -> str: + """Return the one-time secret only to the local foreground launcher.""" + with self._lock: + if self._bootstrap_secret is None: + raise SessionRejected() + return self._bootstrap_secret + + def exchange(self, supplied: str) -> SessionView: + if not isinstance(supplied, str) or len(supplied) > 256: + raise SessionRejected() + with self._lock: + now = self._clock() + expected = self._bootstrap_secret + if ( + expected is None + or now >= self._bootstrap_deadline + or self._bootstrap_failures >= BOOTSTRAP_MAX_FAILURES + or not hmac.compare_digest(expected, supplied) + ): + self._bootstrap_failures += 1 + if self._bootstrap_failures >= BOOTSTRAP_MAX_FAILURES: + self._bootstrap_secret = None + raise SessionRejected() + self._bootstrap_secret = None + token = secrets.token_urlsafe(32) + absolute_expires_at = self._wall_clock().astimezone(timezone.utc) + timedelta( + seconds=SESSION_ABSOLUTE_TTL_SECONDS + ) + session = _Session( + token=token, + idle_deadline=now + SESSION_IDLE_TTL_SECONDS, + absolute_deadline=now + SESSION_ABSOLUTE_TTL_SECONDS, + absolute_expires_at=absolute_expires_at, + ) + self._sessions[token] = session + return SessionView(token=token, expires_at=absolute_expires_at) + + def authorize(self, supplied: str) -> SessionView: + if not isinstance(supplied, str) or len(supplied) > 256: + raise SessionRejected() + with self._lock: + now = self._clock() + session = self._sessions.get(supplied) + if ( + session is None + or not hmac.compare_digest(session.token, supplied) + or now >= session.idle_deadline + or now >= session.absolute_deadline + ): + if session is not None: + self._sessions.pop(supplied, None) + raise SessionRejected() + session.idle_deadline = min( + now + SESSION_IDLE_TTL_SECONDS, session.absolute_deadline + ) + remaining = min(session.idle_deadline, session.absolute_deadline) - now + expires_at = self._wall_clock().astimezone(timezone.utc) + timedelta( + seconds=max(0.0, remaining) + ) + return SessionView(token=session.token, expires_at=expires_at) + + def clear(self) -> None: + with self._lock: + self._bootstrap_secret = None + self._sessions.clear() diff --git a/tests/test_console_server.py b/tests/test_console_server.py new file mode 100644 index 0000000..6fd0804 --- /dev/null +++ b/tests/test_console_server.py @@ -0,0 +1,227 @@ +from __future__ import annotations + +from http.client import HTTPConnection +import json +import socket +from threading import Thread +import unittest + +from dyro.console.server import create_console_http_server + + +class ConsoleServerTests(unittest.TestCase): + def setUp(self) -> None: + self.server = create_console_http_server(port=0, bootstrap_secret="a" * 43) + self.thread = Thread(target=self.server.serve_forever, daemon=True) + self.thread.start() + self.port = self.server.server_port + self.origin = f"http://127.0.0.1:{self.port}" + + def tearDown(self) -> None: + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=5) + + def _request( + self, + method: str, + path: str, + *, + body: bytes | None = None, + headers: dict[str, str] | None = None, + ) -> tuple[int, dict[str, str], bytes]: + connection = HTTPConnection("127.0.0.1", self.port, timeout=5) + connection.request(method, path, body=body, headers=headers or {}) + response = connection.getresponse() + payload = response.read() + result = response.status, dict(response.getheaders()), payload + connection.close() + return result + + def _exchange(self, secret: str = "a" * 43) -> str: + status, headers, body = self._request( + "POST", + "/api/v1/session", + body=json.dumps({"bootstrap": secret}).encode("utf-8"), + headers={ + "Content-Type": "application/json", + "Origin": self.origin, + "Sec-Fetch-Site": "same-origin", + }, + ) + self.assertEqual(status, 201) + self.assertEqual(headers.get("Set-Cookie"), None) + return json.loads(body)["bearer"] + + def test_static_shell_has_no_project_data_or_bootstrap_secret(self) -> None: + status, headers, body = self._request("GET", "/") + + self.assertEqual(status, 200) + self.assertEqual(headers["Content-Security-Policy"], "default-src 'none'; script-src 'self'; style-src 'self'; img-src 'self'; connect-src 'self'; worker-src 'none'; object-src 'none'; base-uri 'none'; form-action 'none'; frame-ancestors 'none'") + self.assertEqual(headers["Cross-Origin-Opener-Policy"], "same-origin") + self.assertEqual(headers["Cache-Control"], "no-store") + self.assertNotIn("a" * 43, body.decode("utf-8")) + self.assertNotIn("dyro.toml", body.decode("utf-8")) + + def test_bootstrap_is_same_origin_single_use_and_issues_independent_bearer(self) -> None: + rejected, _, _ = self._request( + "POST", + "/api/v1/session", + body=json.dumps({"bootstrap": "a" * 43}).encode("utf-8"), + headers={"Content-Type": "application/json", "Origin": "http://example.test"}, + ) + self.assertEqual(rejected, 403) + + bearer = self._exchange() + self.assertNotEqual(bearer, "a" * 43) + replayed, _, _ = self._request( + "POST", + "/api/v1/session", + body=json.dumps({"bootstrap": "a" * 43}).encode("utf-8"), + headers={"Content-Type": "application/json", "Origin": self.origin}, + ) + self.assertEqual(replayed, 401) + + def test_api_requires_exact_host_authorization_and_origin(self) -> None: + unauthorized, _, body = self._request("GET", "/api/v1/meta") + self.assertEqual(unauthorized, 401) + self.assertEqual(json.loads(body)["error"]["code"], "UNAUTHORIZED") + + bearer = self._exchange() + spoofed, _, _ = self._request( + "GET", + "/api/v1/meta", + headers={"Host": "localhost", "Authorization": f"Bearer {bearer}"}, + ) + self.assertEqual(spoofed, 400) + cross_origin, _, _ = self._request( + "GET", + "/api/v1/meta", + headers={"Authorization": f"Bearer {bearer}", "Origin": "http://example.test"}, + ) + self.assertEqual(cross_origin, 403) + + status, headers, body = self._request( + "GET", + "/api/v1/meta", + headers={"Authorization": f"Bearer {bearer}", "Origin": self.origin}, + ) + self.assertEqual(status, 200) + self.assertEqual(headers["Content-Type"], "application/json; charset=utf-8") + payload = json.loads(body) + self.assertEqual(payload["schema_version"], 1) + self.assertEqual(payload["data"]["capabilities"], []) + self.assertIn("session_expires_at", payload["data"]) + + def test_api_refuses_cors_preflight_mutations_and_invalid_request_framing(self) -> None: + for method in ("OPTIONS", "PUT", "DELETE", "PATCH"): + status, headers, _ = self._request(method, "/api/v1/meta") + self.assertEqual(status, 405) + self.assertNotIn("Access-Control-Allow-Origin", headers) + + spoofed, _, _ = self._request( + "OPTIONS", "/api/v1/meta", headers={"Host": "localhost"} + ) + self.assertEqual(spoofed, 400) + + body = json.dumps({"bootstrap": "a" * 43}).encode("utf-8") + status, _, _ = self._request( + "POST", + "/api/v1/session", + body=body, + headers={ + "Content-Type": "application/json", + "Origin": self.origin, + "Transfer-Encoding": "chunked", + }, + ) + self.assertEqual(status, 400) + + def test_server_uses_loopback_only_and_rejects_invalid_ports(self) -> None: + self.assertEqual(self.server.server_address[0], "127.0.0.1") + with self.assertRaises(ValueError): + create_console_http_server(port=-1) + with self.assertRaises(ValueError): + create_console_http_server(port=65536) + + def test_request_line_limit_fails_closed_before_stdlib_parsing(self) -> None: + client = socket.create_connection(("127.0.0.1", self.port), timeout=5) + try: + client.sendall(b"GET /" + b"a" * 5000 + b" HTTP/1.1\r\nHost: ignored\r\n\r\n") + chunks: list[bytes] = [] + while True: + chunk = client.recv(4096) + if not chunk: + break + chunks.append(chunk) + finally: + client.close() + response = b"".join(chunks) + + self.assertIn(b" 400 ", response) + self.assertIn(b"Connection: close", response) + + def test_stdlib_parse_errors_are_sanitized(self) -> None: + client = socket.create_connection(("127.0.0.1", self.port), timeout=5) + try: + client.sendall(b"GET / HTTP/1.1\r\nBroken-Header\r\n\r\n") + chunks: list[bytes] = [] + while True: + chunk = client.recv(4096) + if not chunk: + break + chunks.append(chunk) + finally: + client.close() + response = b"".join(chunks) + + self.assertIn(b" 400 ", response) + self.assertIn(b'"code":"BAD_REQUEST"', response) + self.assertNotIn(b"Broken-Header", response) + + def test_header_limits_and_obs_fold_fail_before_application_routing(self) -> None: + cases = ( + b"GET / HTTP/1.1\r\nHost: ignored\r\n X-Folded: value\r\n\r\n", + b"GET / HTTP/1.1\r\nHost: ignored\r\nX-Large: " + b"a" * 5000 + b"\r\n\r\n", + ) + for request in cases: + with self.subTest(request=request[:24]): + client = socket.create_connection(("127.0.0.1", self.port), timeout=5) + try: + client.sendall(request) + chunks: list[bytes] = [] + while True: + chunk = client.recv(4096) + if not chunk: + break + chunks.append(chunk) + finally: + client.close() + response = b"".join(chunks) + self.assertIn(b"BAD_REQUEST", response) + + def test_each_response_closes_a_pipelined_http11_connection(self) -> None: + client = socket.create_connection(("127.0.0.1", self.port), timeout=5) + request = ( + f"GET /api/v1/meta HTTP/1.1\r\nHost: 127.0.0.1:{self.port}\r\n\r\n" + f"GET / HTTP/1.1\r\nHost: 127.0.0.1:{self.port}\r\n\r\n" + ).encode("ascii") + try: + client.sendall(request) + chunks: list[bytes] = [] + while True: + chunk = client.recv(4096) + if not chunk: + break + chunks.append(chunk) + finally: + client.close() + response = b"".join(chunks) + + self.assertIn(b" 401 ", response) + self.assertIn(b"Connection: close", response) + self.assertNotIn(b"

Dyro Console

", response) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_console_session.py b/tests/test_console_session.py new file mode 100644 index 0000000..5e18219 --- /dev/null +++ b/tests/test_console_session.py @@ -0,0 +1,55 @@ +from __future__ import annotations + +from datetime import datetime, timezone +import unittest + +from dyro.console.session import ( + BOOTSTRAP_MAX_FAILURES, + BOOTSTRAP_TTL_SECONDS, + SESSION_IDLE_TTL_SECONDS, + ConsoleSessionStore, + SessionRejected, +) + + +class ConsoleSessionStoreTests(unittest.TestCase): + def setUp(self) -> None: + self.now = [0.0] + self.wall = [datetime(2026, 8, 4, 12, 0, tzinfo=timezone.utc)] + self.store = ConsoleSessionStore( + bootstrap_secret="a" * 43, + monotonic_clock=lambda: self.now[0], + wall_clock=lambda: self.wall[0], + ) + + def test_bootstrap_expiry_and_failure_limit_fail_closed(self) -> None: + self.now[0] = BOOTSTRAP_TTL_SECONDS + with self.assertRaises(SessionRejected): + self.store.exchange("a" * 43) + + limited = ConsoleSessionStore( + bootstrap_secret="b" * 43, + monotonic_clock=lambda: 0.0, + wall_clock=lambda: self.wall[0], + ) + for _ in range(BOOTSTRAP_MAX_FAILURES): + with self.assertRaises(SessionRejected): + limited.exchange("c" * 43) + with self.assertRaises(SessionRejected): + limited.exchange("b" * 43) + + def test_authorization_refreshes_idle_only_and_remains_memory_local(self) -> None: + session = self.store.exchange("a" * 43) + with self.assertRaises(SessionRejected): + self.store.bootstrap_secret + + self.now[0] = SESSION_IDLE_TTL_SECONDS - 1 + refreshed = self.store.authorize(session.token) + self.assertGreater(refreshed.expires_at, self.wall[0]) + self.now[0] += SESSION_IDLE_TTL_SECONDS + 1 + with self.assertRaises(SessionRejected): + self.store.authorize(session.token) + + +if __name__ == "__main__": + unittest.main()