diff --git a/README.md b/README.md index 081a21e4..0b3fe529 100644 --- a/README.md +++ b/README.md @@ -829,6 +829,8 @@ docker stop cloak && docker rm cloak > **Security:** CDP gives full control over the browser (execute JS, read pages, access files). > The examples bind to `127.0.0.1` so only your machine can connect. Never expose port 9222 > to the public internet without additional authentication. +> In containers, `cloakserve` listens on `0.0.0.0` internally so Docker port publishing works; +> if the service is reachable on all interfaces, startup logs and `GET /` diagnostics include a warning. ### Docker Compose diff --git a/bin/cloakserve b/bin/cloakserve index c51242ea..d4ea50a8 100755 --- a/bin/cloakserve +++ b/bin/cloakserve @@ -364,6 +364,21 @@ def parse_connection_params(query_string: str) -> dict: # HTTP handlers # --------------------------------------------------------------------------- +PUBLIC_BIND_SECURITY_WARNING = ( + "cloakserve is listening on 0.0.0.0 without authentication. " + "CDP grants full browser control; publish port 9222 only to trusted " + "interfaces or place it behind an authenticated reverse proxy." +) + + +def _bind_security_warnings(host: str) -> list[str]: + """Return diagnostics warnings for unsafe cloakserve bind settings.""" + normalized = host.strip().lower() + if normalized in {"0.0.0.0", "::", "[::]"}: + return [PUBLIC_BIND_SECURITY_WARNING] + return [] + + def _ws_scheme(request: web.Request) -> str: """Return 'wss' if client connected via HTTPS (e.g. TLS-terminating proxy), else 'ws'.""" proto = request.headers.get("X-Forwarded-Proto", request.scheme) @@ -385,11 +400,15 @@ async def handle_root(request: web.Request) -> web.Response: "locale": proc.locale, "proxy": proc.proxy, } - return web.json_response({ + response = { "status": "ok", "active": len(processes), "processes": processes, - }) + } + warnings = _bind_security_warnings(request.app.get("host", "")) + if warnings: + response["warnings"] = warnings + return web.json_response(response) async def handle_json_version(request: web.Request) -> web.Response: @@ -661,6 +680,10 @@ def main() -> None: app["pool"] = pool app["port"] = config["port"] + in_container = os.path.exists("/.dockerenv") or os.path.exists("/run/.containerenv") + host = "0.0.0.0" if in_container else "127.0.0.1" + app["host"] = host + # Routes app.router.add_get("/", handle_root) app.router.add_get("/json/version", handle_json_version) @@ -685,8 +708,8 @@ def main() -> None: port, ) - in_container = os.path.exists("/.dockerenv") or os.path.exists("/run/.containerenv") - host = "0.0.0.0" if in_container else "127.0.0.1" + for warning in _bind_security_warnings(host): + logger.warning(warning) web.run_app(app, host=host, port=port, print=None) diff --git a/tests/test_cloakserve.py b/tests/test_cloakserve.py index a1dccb16..b4976a39 100644 --- a/tests/test_cloakserve.py +++ b/tests/test_cloakserve.py @@ -1,9 +1,12 @@ """Unit tests for cloakserve — parse_connection_params, parse_cli_args, URL rewriting, connection tracking.""" +import asyncio import importlib.machinery import importlib.util +import json import sys from pathlib import Path +from types import SimpleNamespace from unittest.mock import patch import pytest @@ -197,6 +200,41 @@ def test_wss_scheme_list(self): assert result == "wss://host:443/fingerprint/seed1/devtools/page/DEF-456" +# --------------------------------------------------------------------------- +# Bind security warnings +# --------------------------------------------------------------------------- + + +class TestBindSecurityWarnings: + """Warn when cloakserve is reachable on all interfaces without auth.""" + + def test_public_ipv4_bind_warns_about_unauthenticated_cdp(self): + warnings = _mod._bind_security_warnings("0.0.0.0") + assert len(warnings) == 1 + assert "without authentication" in warnings[0] + assert "CDP" in warnings[0] + + def test_public_ipv6_bind_warns_about_unauthenticated_cdp(self): + warnings = _mod._bind_security_warnings("::") + assert len(warnings) == 1 + assert "without authentication" in warnings[0] + + def test_loopback_bind_has_no_warning(self): + assert _mod._bind_security_warnings("127.0.0.1") == [] + assert _mod._bind_security_warnings("localhost") == [] + + def test_root_diagnostics_include_public_bind_warning(self): + request = SimpleNamespace(app={ + "host": "0.0.0.0", + "pool": SimpleNamespace(_processes={}, _connections={}), + }) + + response = asyncio.run(_mod.handle_root(request)) + payload = json.loads(response.text) + + assert payload["warnings"] == _mod._bind_security_warnings("0.0.0.0") + + # --------------------------------------------------------------------------- # Connection refcounting # ---------------------------------------------------------------------------