Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
31 changes: 27 additions & 4 deletions bin/cloakserve
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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:
Expand Down Expand Up @@ -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)
Expand All @@ -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)


Expand Down
38 changes: 38 additions & 0 deletions tests/test_cloakserve.py
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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
# ---------------------------------------------------------------------------
Expand Down