"""Hermes Web UI server entry point."""import loggingimport osimport reimport signalimport socketimport sslimport sysimport threadingimport timeimport tracebackfrom http.server import BaseHTTPRequestHandler, ThreadingHTTPServer # Ignore SIGPIPE so a dropped client only aborts that write, not the whole WebUI process._SIGPIPE = getattr(signal, "SIGPIPE", None)if _SIGPIPE is not None:    signal.signal(_SIGPIPE, signal.SIG_IGN) # Test-mode network isolation keeps subprocess-backed tests hermetic.if os.environ.get("HERMES_WEBUI_TEST_NETWORK_BLOCK", "").strip() in ("1", "true", "yes"):    _REAL_CREATE_CONN = socket.create_connection    _REAL_SOCK_CONNECT = socket.socket.connect     import re as _re     def _re_match_unique_local_ipv6(h):        """Match IPv6 fc00::/7 without catching similar-looking hostnames."""        return bool(_re.match(r"^f[cd][0-9a-f]{0,2}:", h))     def _addr_is_local(host):        if not isinstance(host, str):            return False        h = host.strip().lower()        if not h:            return False        if h in ("::1", "0:0:0:0:0:0:0:1") or h.startswith("fe80:") or _re_match_unique_local_ipv6(h):            return True        if h == "localhost" or h.endswith(".localhost"):            return True        if h.endswith(".local") or h.endswith(".test") or h.endswith(".invalid"):            return True        if h == "example.com" or h.endswith(".example.com"):            return True        if h == "example.net" or h.endswith(".example.net"):            return True        if h == "example.org" or h.endswith(".example.org"):            return True        if h.endswith(".example"):            return True        if h and h[0].isdigit() and h.count(".") == 3:            try:                o1, o2, o3, o4 = [int(p) for p in h.split(".")]            except ValueError:                return False            if o1 == 127:                return True            if o1 == 10:                return True            if o1 == 192 and o2 == 168:                return True            if o1 == 172 and 16 <= o2 <= 31:                return True            if o1 == 169 and o2 == 254:                return True            if o1 == 203 and o2 == 0 and o3 == 113:                return True        return False     def _blocked_create_connection(address, *a, **kw):        try:            host = address[0]        except (TypeError, IndexError):            host = ""        if _addr_is_local(host):            return _REAL_CREATE_CONN(address, *a, **kw)        raise OSError(            f"hermes test network isolation (server.py): outbound to {address!r} blocked"        )     def _blocked_socket_connect(self, address):        try:            host = address[0]        except (TypeError, IndexError):            host = ""        if _addr_is_local(host):            return _REAL_SOCK_CONNECT(self, address)        raise OSError(            f"hermes test network isolation (server.py): socket.connect to {address!r} blocked"        )     socket.create_connection = _blocked_create_connection    socket.socket.connect = _blocked_socket_connect  try:    import resourceexcept ImportError:  # pragma: no cover - resource is Unix-only    resource = Nonefrom urllib.parse import urlparse logger = logging.getLogger(__name__) from api.auth import check_auth, reset_trusted_auth_request_statefrom api.config import HOST, PORT, STATE_DIR, SESSION_DIR, DEFAULT_WORKSPACEfrom api.helpers import (    j,    get_profile_cookie,    _build_csp_report_only_policy,    _CLIENT_DISCONNECT_ERRORS,)from api.profiles import set_request_profile, clear_request_profilefrom api.routes import handle_delete, handle_get, handle_patch, handle_post, handle_put, apply_cors_preflight_headersfrom api.startup import auto_install_agent_deps, fix_credential_permissionsfrom api.updates import WEBUI_VERSIONfrom api.crash_visibility import install_crash_visibility  class QuietHTTPServer(ThreadingHTTPServer):    """Custom HTTP server that silently handles common network errors."""    daemon_threads = True    request_queue_size = 64    max_request_workers = 128    max_overflow_reject_workers = 16    _OVERFLOW_RESPONSE = (        b"HTTP/1.1 503 Service Unavailable\r\n"        b"Connection: close\r\n"        b"Content-Length: 0\r\n"        b"\r\n"    )     def __init__(self, *args, **kwargs):        server_address = args[0] if args else kwargs.get('server_address', None)        if server_address and ':' in server_address[0]:            self.address_family = socket.AF_INET6        self.ssl_context: object | None = None        super().__init__(*args, **kwargs)        self._request_worker_slots = threading.BoundedSemaphore(self.max_request_workers)        self._overflow_reject_slots = threading.BoundedSemaphore(self.max_overflow_reject_workers)        self.accept_loop_requests_total = 0        self.accept_loop_last_request_at = 0.0     def server_bind(self):        if sys.platform == 'win32':            self.allow_reuse_address = False            SO_EXCLUSIVEADDRUSE = getattr(socket, 'SO_EXCLUSIVEADDRUSE', -5)            self.socket.setsockopt(socket.SOL_SOCKET, SO_EXCLUSIVEADDRUSE, 1)            # Retry bind on Windows to handle the case where a previous            # process (e.g. during self-update) is still releasing the port.            # The old process calls os._exit(0) which starts tearing down            # its socket, but with SO_EXCLUSIVEADDRUSE the OS blocks new            # binds until the teardown completes.  Retry for up to 10 s.            max_retries = 20            retry_delay = 0.5            for attempt in range(max_retries):                try:                    super().server_bind()                    return                except OSError as e:                    if e.winerror == 10048 and attempt < max_retries - 1:  # WSAEADDRINUSE                        time.sleep(retry_delay)                    else:                        raise        else:            super().server_bind()     def get_request(self):        """Accept raw sockets and defer TLS handshake work to request threads."""        request, client_address = self.socket.accept()        ssl_context = getattr(self, "ssl_context", None)        if ssl_context is None:            return request, client_address        try:            tls_request = ssl_context.wrap_socket(                request,                server_side=True,                do_handshake_on_connect=False,            )        except Exception:            request.close()            raise        return tls_request, client_address     def _handle_request_noblock(self):        """Record accept-loop progress before dispatching a request handler."""        self.accept_loop_requests_total += 1        self.accept_loop_last_request_at = time.time()        return super()._handle_request_noblock()     def _close_request_quietly(self, request) -> None:        try:            request.close()        except Exception:            pass     def _drain_request_input_nonblocking(self, request) -> None:        # Read through the current header block before replying so Windows        # doesn't reset the socket when we close with unread input.        deadline = time.monotonic() + 0.05        buffered = bytearray()        header_terminator = b"\r\n\r\n"        max_bytes = 65536        try:            timeout = request.gettimeout()        except Exception:            timeout = None        try:            while len(buffered) < max_bytes and header_terminator not in buffered:                wait = deadline - time.monotonic()                if wait <= 0:                    break                try:                    request.settimeout(wait)                    chunk = request.recv(min(4096, max_bytes - len(buffered)))                except (BlockingIOError, InterruptedError, TimeoutError, socket.timeout):                    break                except OSError:                    break                if not chunk:                    break                buffered.extend(chunk)            try:                request.shutdown(socket.SHUT_RD)            except OSError:                pass        finally:            try:                request.settimeout(timeout)            except Exception:                pass     def _reject_overflow_request(self, request) -> None:        if getattr(self, "ssl_context", None) is not None:            self._close_request_quietly(request)            return        if not self._overflow_reject_slots.acquire(blocking=False):            self._close_request_quietly(request)            return        try:            threading.Thread(                target=self._reject_overflow_request_worker,                args=(request,),                daemon=True,            ).start()        except Exception:            self._overflow_reject_slots.release()            self._close_request_quietly(request)     def _reject_overflow_request_worker(self, request) -> None:        try:            self._drain_request_input_nonblocking(request)            try:                request.sendall(self._OVERFLOW_RESPONSE)                try:                    request.shutdown(socket.SHUT_WR)                except Exception:                    pass            except Exception:                pass        finally:            self._close_request_quietly(request)            self._overflow_reject_slots.release()     def process_request(self, request, client_address):        if not self._request_worker_slots.acquire(blocking=False):            self._reject_overflow_request(request)            return        try:            return super().process_request(request, client_address)        except Exception:            self._request_worker_slots.release()            self._close_request_quietly(request)            raise     def process_request_thread(self, request, client_address):        try:            return super().process_request_thread(request, client_address)        finally:            self._request_worker_slots.release()     def handle_error(self, request, client_address):        """Suppress logging for common client disconnect errors."""        exc_type, exc_value, _ = sys.exc_info()        if exc_type in (            ConnectionResetError, BrokenPipeError, ConnectionAbortedError,            TimeoutError, ssl.SSLError, ssl.SSLEOFError,        ):            return        if issubclass(exc_type, OSError):            if getattr(exc_value, 'errno', None) in (32, 54, 104, 110):  # EPIPE, ECONNRESET, ETIMEDOUT                return        super().handle_error(request, client_address)  class Handler(BaseHTTPRequestHandler):    # HTTP/1.1 keep-alive stays on, so every response must declare framing.    protocol_version = "HTTP/1.1"    timeout = 30  # seconds — kills idle/incomplete connections to prevent thread exhaustion        def setup(self):        """Set socket options for each accepted connection."""        super().setup()        try:            self.connection.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)        except OSError:            pass        try:            self.connection.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)        except OSError:            pass        if hasattr(socket, 'TCP_KEEPIDLE'):  # Linux            try:                self.connection.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 10)                self.connection.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 5)                self.connection.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, 3)            except OSError:                pass        elif hasattr(socket, 'TCP_KEEPALIVE'):  # macOS            try:                self.connection.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPALIVE, 10)            except OSError:                pass    _ver_suffix = WEBUI_VERSION.removeprefix('v')    server_version = ('HermesWebUI/' + _ver_suffix) if _ver_suffix != 'unknown' else 'HermesWebUI'    _CSP_REPORT_TO = '{"group":"csp-endpoint","max_age":10886400,"endpoints":[{"url":"/api/csp-report"}]}'     @classmethod    def csp_report_only_policy(cls, extra_connect_src=None, extra_frame_src=None) -> str:        return _build_csp_report_only_policy(extra_connect_src, extra_frame_src)     def end_headers(self) -> None:        extra_connect_src = getattr(self, "_csp_extra_connect_src", None)        extra_frame_src = getattr(self, "_csp_extra_frame_src", None)        self.send_header("Content-Security-Policy-Report-Only", self.csp_report_only_policy(extra_connect_src, extra_frame_src))        self.send_header("Report-To", self._CSP_REPORT_TO)        super().end_headers()     def log_message(self, fmt, *args): pass  # suppress default Apache-style log     @staticmethod    def _safe_webui_print(message: str) -> None:        """Emit a request log line without letting logging break responses."""        try:            print(message, flush=True)        except Exception:            pass     def log_request(self, code: str='-', size: str='-') -> None:        """Structured JSON logs for each request."""        import json as _json        duration_ms = round((time.time() - getattr(self, '_req_t0', time.time())) * 1000, 1)        remote = '-'        try:            if getattr(self, 'client_address', None):                remote = str(self.client_address[0])        except Exception:            remote = '-'        forwarded_for = None        try:            forwarded_for = (self.headers.get('X-Forwarded-For') or '').split(',')[0].strip() or None        except Exception:            forwarded_for = None        record_data = {            'ts': time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()),            'remote': remote,            'method': getattr(self, 'command', None) or '-',            'path': getattr(self, 'path', None) or '-',            'status': int(code) if str(code).isdigit() else code,            'ms': duration_ms,        }        if forwarded_for:            record_data['forwarded_for'] = forwarded_for        record = _json.dumps(record_data)        self._safe_webui_print(f'[webui] {record}')     def do_GET(self) -> None:        self._req_t0 = time.time(); reset_trusted_auth_request_state(self)        cookie_profile = get_profile_cookie(self)        if cookie_profile:            set_request_profile(cookie_profile)        try:            parsed = urlparse(self.path)            if not check_auth(self, parsed): return            result = handle_get(self, parsed)            if result is False:                return j(self, {'error': 'not found'}, status=404)        except _CLIENT_DISCONNECT_ERRORS:            # Expected disconnect path; do not convert it into a misleading server 500.            return        except Exception:            self._safe_webui_print(f'[webui] ERROR {self.command} {self.path}\n' + traceback.format_exc())            try:                j(self, {'error': 'Internal server error'}, status=500)            except _CLIENT_DISCONNECT_ERRORS:                pass            except Exception:                self._safe_webui_print(traceback.format_exc())        finally:            clear_request_profile()     def _handle_write(self, route_func) -> None:        self._req_t0 = time.time(); reset_trusted_auth_request_state(self)        cookie_profile = get_profile_cookie(self)        if cookie_profile:            set_request_profile(cookie_profile)        try:            parsed = urlparse(self.path)            _is_csp_report_post = (                parsed.path == "/api/csp-report" and self.command == "POST"            )            if not _is_csp_report_post and not check_auth(self, parsed): return            result = route_func(self, parsed)            if result is False:                return j(self, {'error': 'not found'}, status=404)        except _CLIENT_DISCONNECT_ERRORS:            # Expected disconnect path; do not convert it into a misleading server 500.            return        except Exception:            self._safe_webui_print(f'[webui] ERROR {self.command} {self.path}\n' + traceback.format_exc())            try:                j(self, {'error': 'Internal server error'}, status=500)            except _CLIENT_DISCONNECT_ERRORS:                pass            except Exception:                self._safe_webui_print(traceback.format_exc())        finally:            clear_request_profile()     def do_POST(self) -> None:        self._handle_write(handle_post)     def do_PUT(self) -> None:        self._handle_write(handle_put)     def do_PATCH(self) -> None:        self._handle_write(handle_patch)     def do_OPTIONS(self) -> None:        """Handle CORS preflight requests (headers emitted by api.routes)."""        self._req_t0 = time.time()        self.send_response(200)        apply_cors_preflight_headers(self)        # Frame the empty preflight: without Content-Length an HTTP/1.1 keep-alive        # 200 is read-until-close, hanging the client until the 30s timeout.        self.send_header("Content-Length", "0")        self.end_headers()     def do_DELETE(self) -> None:        self._handle_write(handle_delete)  def _raise_fd_soft_limit(target: int = 4096) -> dict:    """Best-effort raise of RLIMIT_NOFILE for persistent WebUI hosts."""    if resource is None:        return {"status": "unsupported"}    try:        soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE)    except Exception as exc:        return {"status": "error", "error": str(exc)}     desired = int(target)    if hard not in (-1, getattr(resource, "RLIM_INFINITY", object())):        desired = min(desired, int(hard))    if soft >= desired:        return {"status": "unchanged", "soft": soft, "hard": hard}    try:        resource.setrlimit(resource.RLIMIT_NOFILE, (desired, hard))    except Exception as exc:        return {"status": "error", "soft": soft, "hard": hard, "error": str(exc)}    return {"status": "raised", "soft": desired, "hard": hard, "previous_soft": soft}  _SHUTDOWN_AUDIT_LOGGED = False_SHUTDOWN_LOG_VALUE_RE = re.compile(r"[\x00-\x1f\x7f]+")  def _shutdown_log_value(value, *, default: str = "unknown", max_len: int = 160) -> str:    """Return a bounded single-line value safe for shutdown diagnostics."""    if value is None:        return default    try:        text = str(value)    except Exception:        return default    text = _SHUTDOWN_LOG_VALUE_RE.sub("?", text).strip()    if not text:        return default    if len(text) > max_len:        text = f"{text[:max_len]}…"    return text  def _log_shutdown_audit(reason: str = "serve_forever_exit") -> None:    """Log runtime context when the WebUI server is exiting."""    global _SHUTDOWN_AUDIT_LOGGED    if _SHUTDOWN_AUDIT_LOGGED:        return     active_sessions = []    try:        from api.models import LOCK, SESSIONS

Lines 1–500 of 749. Load more