From 01d152b06f558ec4b63a62d4ea04b5ff28579ecc Mon Sep 17 00:00:00 2001 From: oobabooga Date: Fri, 12 Jun 2026 14:36:35 -0300 Subject: [PATCH] Studio: only advertise a Cloudflare tunnel once it actually serves (#6264) --- studio/backend/cloudflare_tunnel.py | 159 ++++++--- studio/backend/run.py | 6 +- .../backend/tests/test_cloudflare_tunnel.py | 304 +++++++++++++++++- 3 files changed, 415 insertions(+), 54 deletions(-) diff --git a/studio/backend/cloudflare_tunnel.py b/studio/backend/cloudflare_tunnel.py index 7c088b6e1e..e5dba69452 100644 --- a/studio/backend/cloudflare_tunnel.py +++ b/studio/backend/cloudflare_tunnel.py @@ -24,12 +24,20 @@ from pathlib import Path from typing import Optional, Tuple # cloudflared logs the quick-tunnel URL; match only the URL so we do not depend -# on the surrounding wording, which Cloudflare may change. -_URL_RE = re.compile(r"https://[A-Za-z0-9-]+\.trycloudflare\.com") +# on the surrounding wording, which Cloudflare may change. The negative lookahead +# drops cloudflared's own API host, which appears in failure lines such as +# failed to request quick Tunnel: Post "https://api.trycloudflare.com/tunnel" +# and must never be mistaken for a usable tunnel URL. +_URL_RE = re.compile(r"https://(?!api\.)[A-Za-z0-9-]+\.trycloudflare\.com") + +# cloudflared logs this once per edge connection it establishes. Until at least +# one appears the quick-tunnel URL returns Cloudflare error 1033 (HTTP 530), so +# we wait for it before advertising the URL. +_REGISTERED_MARKER = "Registered tunnel connection" _RELEASE_BASE = "https://github.com/cloudflare/cloudflared/releases/latest/download" -_URL_TIMEOUT = 15.0 # seconds to wait for the public URL before giving up +_READY_TIMEOUT = 15.0 # seconds to wait for the URL + a registered edge connection _DOWNLOAD_TIMEOUT = 60 # urlopen timeout for the one-time binary download @@ -180,13 +188,24 @@ class CloudflareTunnel: upstream stays local-only. """ - def __init__(self, port: int, binary: str): + def __init__( + self, + port: int, + binary: str, + protocol: Optional[str] = None, + ): self.port = port self.binary = binary + # None lets cloudflared pick its default (quic, with its own http2 + # fallback); set to "http2" to force it when quic is blocked. + self.protocol = protocol self._proc: Optional[subprocess.Popen] = None self._lock = threading.Lock() + self._stopped = False self._url_event = threading.Event() + self._ready_event = threading.Event() self.url: Optional[str] = None + self.ready = False self.error: Optional[str] = None def start(self) -> None: @@ -197,25 +216,33 @@ class CloudflareTunnel: f"http://localhost:{self.port}", "--no-autoupdate", ] - proc = subprocess.Popen( - cmd, - stdout = subprocess.PIPE, - stderr = subprocess.STDOUT, - stdin = subprocess.DEVNULL, - text = True, - errors = "replace", - bufsize = 1, - **_windows_hidden_kwargs(), - ) + if self.protocol: + cmd += ["--protocol", self.protocol] with self._lock: + # A stop() that landed before us (e.g. a shutdown in the caller's + # register->start window) marks the tunnel stopped; spawning now would + # orphan a process nobody owns, so refuse. + if self._stopped: + return + proc = subprocess.Popen( + cmd, + stdout = subprocess.PIPE, + stderr = subprocess.STDOUT, + stdin = subprocess.DEVNULL, + text = True, + errors = "replace", + bufsize = 1, + **_windows_hidden_kwargs(), + ) self._proc = proc threading.Thread( target = self._reader, args = (proc,), name = "cloudflared-reader", daemon = True ).start() def _reader(self, proc: subprocess.Popen) -> None: - # Drain cloudflared's output, capture the first trycloudflare URL, and - # keep draining so it never blocks on a full pipe. + # Drain cloudflared's output: capture the first trycloudflare URL and the + # first edge-connection registration, and keep draining so it never + # blocks on a full pipe. try: if proc.stdout is not None: for line in proc.stdout: @@ -224,20 +251,35 @@ class CloudflareTunnel: if match: self.url = match.group(0) self._url_event.set() + if not self.ready and _REGISTERED_MARKER in line: + self.ready = True + self._ready_event.set() except Exception: pass finally: + # stdout closed -> cloudflared has exited. Record why, and unblock any + # waiters at once instead of letting them wait out the full timeout. if self.url is None: self.error = "cloudflared exited before emitting a tunnel URL" - self._url_event.set() + elif not self.ready: + self.error = "cloudflared exited before the tunnel connection registered" + self._url_event.set() + self._ready_event.set() - def wait_for_url(self, timeout: float = _URL_TIMEOUT) -> Optional[str]: - self._url_event.wait(timeout) - return self.url + def wait_for_ready(self, timeout: float = _READY_TIMEOUT) -> Optional[str]: + """Block until the tunnel is actually serving -- the URL has been minted + *and* at least one edge connection has registered -- or until timeout. + + Returns the URL only when ready, so callers never advertise a URL that + would return Cloudflare error 1033 (HTTP 530).""" + self._ready_event.wait(timeout) + return self.url if self.ready else None def stop(self) -> None: """Terminate the tunnel. Idempotent and safe to call from a signal handler.""" with self._lock: + # Mark stopped so a start() racing behind us refuses to spawn. + self._stopped = True proc, self._proc = self._proc, None if proc is None: return @@ -260,43 +302,74 @@ class CloudflareTunnel: # enough; the lock guards the start/stop/shutdown races. _active_tunnel: Optional[CloudflareTunnel] = None _active_lock = threading.Lock() +# Latched by stop_studio_tunnel so a shutdown landing *between* a start's retry +# attempts aborts the loop instead of starting a tunnel nobody will ever stop. +_shutdown_requested = False -def start_studio_tunnel(port: int, timeout: float = _URL_TIMEOUT) -> Optional[str]: - """Start a quick tunnel and return its public URL, or None (best-effort). +def start_studio_tunnel(port: int, timeout: float = _READY_TIMEOUT) -> Optional[str]: + """Start a quick tunnel and return its public URL once it is actually + serving, or None (best-effort). - On any failure (no binary, no URL within timeout, early crash) the tunnel is - stopped and None is returned, so the caller prints a hint and continues. + Waits for cloudflared to both mint the URL and register an edge connection + before returning, so the caller never advertises a URL that yields Cloudflare + error 1033 (HTTP 530). If a URL is minted but no connection registers within + the window (e.g. quic is blocked on this network), retries once forcing the + http2 protocol. On any failure the tunnel is stopped and None is returned. """ - global _active_tunnel + global _active_tunnel, _shutdown_requested binary = ensure_cloudflared() if not binary: return None - tunnel = CloudflareTunnel(port, binary) - # Register before start/wait so a shutdown during the URL wait can stop it. with _active_lock: - prior, _active_tunnel = _active_tunnel, tunnel - if prior is not None: - prior.stop() - try: - tunnel.start() - url = tunnel.wait_for_url(timeout) - except Exception: - url = None - if url: - return url - # No URL (or crash): drop it unless a concurrent shutdown already replaced it. - with _active_lock: - if _active_tunnel is tunnel: - _active_tunnel = None - tunnel.stop() + _shutdown_requested = False # fresh session + # Default protocol first (quic, with cloudflared's own http2 fallback); if a + # URL appears but no connection registers, quic is likely blocked -> retry + # once forcing http2. + for protocol in (None, "http2"): + # Create + register under the lock, and bail if a stop already landed + # (e.g. between this and the previous attempt) so we never start a tunnel + # after shutdown has run. + with _active_lock: + if _shutdown_requested: + _active_tunnel = None + return None + tunnel = CloudflareTunnel(port, binary, protocol = protocol) + prior, _active_tunnel = _active_tunnel, tunnel + if prior is not None: + prior.stop() + try: + tunnel.start() + url = tunnel.wait_for_ready(timeout) + except Exception: + url = None + if url: + return url + saw_url = tunnel.url is not None + # Not ready: drop it, but only if we are still the active tunnel. + with _active_lock: + was_active = _active_tunnel is tunnel + if was_active: + _active_tunnel = None + tunnel.stop() + # A concurrent shutdown or start took over while we waited; retrying would + # spawn a tunnel nobody owns (orphaned after shutdown), so bail instead. + if not was_active: + return None + # No URL at all is an API/network failure, not a protocol one; forcing + # http2 will not help, so do not burn another window on it. + if not saw_url: + return None return None def stop_studio_tunnel() -> None: """Terminate the active tunnel, if any. Idempotent.""" - global _active_tunnel + global _active_tunnel, _shutdown_requested with _active_lock: + # Latch so an in-flight start_studio_tunnel won't start a fresh tunnel + # (e.g. its http2 retry) after we have already torn down. + _shutdown_requested = True tunnel, _active_tunnel = _active_tunnel, None if tunnel is not None: tunnel.stop() diff --git a/studio/backend/run.py b/studio/backend/run.py index a8e9a46a66..84991a71bb 100644 --- a/studio/backend/run.py +++ b/studio/backend/run.py @@ -1032,9 +1032,13 @@ def run_server( _cloudflare_enabled = cloudflare and host == "0.0.0.0" and not api_only and not _IS_COLAB if _cloudflare_enabled: try: # best-effort: any failure must not block startup - from cloudflare_tunnel import start_studio_tunnel + from cloudflare_tunnel import start_studio_tunnel, stop_studio_tunnel + _cloudflare_url = start_studio_tunnel(port) app.state.cloudflare_url = _cloudflare_url + # Backstop: tear the tunnel down even on an abnormal exit that bypasses + # _graceful_shutdown (e.g. an exception after startup -> sys.exit). Idempotent. + atexit.register(stop_studio_tunnel) except Exception as e: logger.debug("Cloudflare tunnel skipped: %s", e) diff --git a/studio/backend/tests/test_cloudflare_tunnel.py b/studio/backend/tests/test_cloudflare_tunnel.py index 873547631d..8042240b64 100644 --- a/studio/backend/tests/test_cloudflare_tunnel.py +++ b/studio/backend/tests/test_cloudflare_tunnel.py @@ -52,6 +52,26 @@ def test_url_regex_no_match_on_unrelated(): assert ct._URL_RE.search("INF connecting to https://api.cloudflare.com/v4") is None +def test_url_regex_ignores_api_endpoint(): + # cloudflared's failure line names its own API host; it must never be taken + # as the tunnel URL (it returns a 404 and is not a quick tunnel). + line = ( + 'failed to request quick Tunnel: Post "https://api.trycloudflare.com/tunnel": ' + "context deadline exceeded" + ) + assert ct._URL_RE.search(line) is None + + +def test_url_regex_skips_api_host_but_matches_real_url(): + blob = ( + 'ERR failed to request quick Tunnel: Post "https://api.trycloudflare.com/tunnel"\n' + "INF | https://brave-mountain-river-clouds.trycloudflare.com |\n" + ) + m = ct._URL_RE.search(blob) + assert m is not None + assert m.group(0) == "https://brave-mountain-river-clouds.trycloudflare.com" + + # ── asset mapping ──────────────────────────────────────────────────── @@ -295,9 +315,88 @@ def test_stop_terminates_process(): t.stop() -def test_wait_for_url_times_out_without_blocking(): +def test_start_after_stop_does_not_spawn(monkeypatch): + # If stop() lands before start() (a concurrent shutdown in the caller's + # register->start window), start() must NOT spawn a cloudflared process -- + # nobody would own it and it would be orphaned. t = ct.CloudflareTunnel(8080, "/bin/cloudflared") - assert t.wait_for_url(timeout = 0.05) is None + spawned = [] + + class _FakeProc: + stdout = None + + def poll(self): + return 0 + + monkeypatch.setattr(ct.subprocess, "Popen", lambda *a, **k: (spawned.append(a), _FakeProc())[1]) + t.stop() # proc is None -> no-op terminate, but marks the tunnel stopped + t.start() # must short-circuit before Popen + assert spawned == [] + assert t._proc is None + + +def test_wait_for_ready_times_out_without_blocking(): + t = ct.CloudflareTunnel(8080, "/bin/cloudflared") + assert t.wait_for_ready(timeout = 0.05) is None + + +def _fake_proc(text): + return types.SimpleNamespace(stdout = io.StringIO(text)) + + +def test_reader_captures_url_and_registration(): + t = ct.CloudflareTunnel(8080, "/bin/cloudflared") + t._reader( + _fake_proc( + "INF Requesting new quick Tunnel on trycloudflare.com...\n" + "INF | https://words-here-abc.trycloudflare.com |\n" + "INF Registered tunnel connection connIndex=0 protocol=http2\n" + ) + ) + assert t.url == "https://words-here-abc.trycloudflare.com" + assert t.ready is True + assert t.wait_for_ready(0) == t.url + assert t.error is None # a fully-registered tunnel records no error + + +def test_reader_url_without_registration_is_not_ready(): + # A URL but no "Registered tunnel connection" (e.g. quic control stream + # fails) must not be advertised -- it returns Cloudflare error 1033. + t = ct.CloudflareTunnel(8080, "/bin/cloudflared") + t._reader( + _fake_proc( + "INF | https://words-here-abc.trycloudflare.com |\n" + 'ERR failed to serve tunnel connection error="control stream failure"\n' + ) + ) + assert t.url == "https://words-here-abc.trycloudflare.com" + assert t.ready is False + assert t.wait_for_ready(0) is None + assert t.error == "cloudflared exited before the tunnel connection registered" + + +def test_reader_handles_none_stdout(): + # Popen.stdout can be None; _reader must not crash and must leave the tunnel + # un-ready so wait_for_ready returns None. + t = ct.CloudflareTunnel(8080, "/bin/cloudflared") + t._reader(types.SimpleNamespace(stdout = None)) + assert t.url is None + assert t.ready is False + assert t.wait_for_ready(0) is None + assert t.error == "cloudflared exited before emitting a tunnel URL" + + +def test_reader_ignores_api_endpoint_failure_line(): + t = ct.CloudflareTunnel(8080, "/bin/cloudflared") + t._reader( + _fake_proc( + "ERR failed to request quick Tunnel: Post " + '"https://api.trycloudflare.com/tunnel": context deadline exceeded\n' + ) + ) + assert t.url is None + assert t.wait_for_ready(0) is None + assert t.error == "cloudflared exited before emitting a tunnel URL" def test_start_studio_tunnel_no_binary(monkeypatch): @@ -306,18 +405,23 @@ def test_start_studio_tunnel_no_binary(monkeypatch): def test_start_studio_tunnel_registers_before_wait(monkeypatch): - # The tunnel must be visible to stop_studio_tunnel() during the URL wait, - # else a shutdown in that window orphans cloudflared. + # The tunnel must be visible to stop_studio_tunnel() during the readiness + # wait, else a shutdown in that window orphans cloudflared. seen = {} class _Stub: - def __init__(self, port, binary): + def __init__( + self, + port, + binary, + protocol = None, + ): self.url = None def start(self): pass - def wait_for_url(self, timeout): + def wait_for_ready(self, timeout): seen["active_during_wait"] = ct._active_tunnel is self self.url = "https://x.trycloudflare.com" return self.url @@ -338,13 +442,18 @@ def test_start_studio_tunnel_clears_and_stops_on_no_url(monkeypatch): seen = {} class _Stub: - def __init__(self, port, binary): + def __init__( + self, + port, + binary, + protocol = None, + ): self.url = None def start(self): pass - def wait_for_url(self, timeout): + def wait_for_ready(self, timeout): return None def stop(self): @@ -359,13 +468,18 @@ def test_start_studio_tunnel_clears_and_stops_on_no_url(monkeypatch): def test_start_studio_tunnel_returns_url(monkeypatch): class _StubTunnel: - def __init__(self, port, binary): + def __init__( + self, + port, + binary, + protocol = None, + ): self.url = None def start(self): self.url = "https://stub-xyz.trycloudflare.com" - def wait_for_url(self, timeout): + def wait_for_ready(self, timeout): return self.url def stop(self): @@ -379,6 +493,169 @@ def test_start_studio_tunnel_returns_url(monkeypatch): ct.stop_studio_tunnel() +def test_start_studio_tunnel_falls_back_to_http2(monkeypatch): + # First attempt mints a URL but never registers (quic blocked); the http2 + # retry registers and wins. + attempts = [] + + class _Stub: + def __init__( + self, + port, + binary, + protocol = None, + ): + self.protocol = protocol + self.url = None + attempts.append(protocol) + + def start(self): + self.url = "https://words.trycloudflare.com" # URL always minted + + def wait_for_ready(self, timeout): + return self.url if self.protocol == "http2" else None + + def stop(self): + pass + + monkeypatch.setattr(ct, "ensure_cloudflared", lambda: "/bin/cloudflared") + monkeypatch.setattr(ct, "CloudflareTunnel", _Stub) + try: + assert ct.start_studio_tunnel(8080) == "https://words.trycloudflare.com" + assert attempts == [None, "http2"] # default first, then forced http2 + finally: + ct.stop_studio_tunnel() + + +def test_start_studio_tunnel_no_retry_when_shutdown_between_attempts(monkeypatch): + # A stop() landing in the gap AFTER the failed first attempt is cleaned up but + # BEFORE the http2 retry registers must abort the loop -- not start a second + # tunnel that nobody will ever stop (Codex review). Simulated by having the + # first attempt's stop() (called during cleanup) trigger the shutdown. + attempts = [] + + class _Stub: + def __init__( + self, + port, + binary, + protocol = None, + ): + self.url = None + attempts.append(protocol) + + def start(self): + self.url = "https://words.trycloudflare.com" # URL minted, never ready + + def wait_for_ready(self, timeout): + return None + + def stop(self): + ct.stop_studio_tunnel() # a concurrent shutdown lands in the gap + + monkeypatch.setattr(ct, "ensure_cloudflared", lambda: "/bin/cloudflared") + monkeypatch.setattr(ct, "CloudflareTunnel", _Stub) + assert ct.start_studio_tunnel(8080) is None + assert attempts == [None] # http2 retry aborted after shutdown + assert ct._active_tunnel is None + + +def test_start_studio_tunnel_no_http2_retry_when_no_url(monkeypatch): + # No URL at all is an API/network failure; the http2 fallback would not help, + # so it must be skipped (don't burn a second timeout window). + attempts = [] + + class _Stub: + def __init__( + self, + port, + binary, + protocol = None, + ): + self.url = None + attempts.append(protocol) + + def start(self): + pass # never mints a URL + + def wait_for_ready(self, timeout): + return None + + def stop(self): + pass + + monkeypatch.setattr(ct, "ensure_cloudflared", lambda: "/bin/cloudflared") + monkeypatch.setattr(ct, "CloudflareTunnel", _Stub) + assert ct.start_studio_tunnel(8080) is None + assert attempts == [None] + + +def test_start_studio_tunnel_both_protocols_fail_registration(monkeypatch): + # Both quic and http2 mint a URL but neither registers -> both attempts are + # exhausted and None is returned (no dead URL advertised). + attempts = [] + + class _Stub: + def __init__( + self, + port, + binary, + protocol = None, + ): + self.url = None + attempts.append(protocol) + + def start(self): + self.url = "https://words.trycloudflare.com" # URL minted, never ready + + def wait_for_ready(self, timeout): + return None + + def stop(self): + pass + + monkeypatch.setattr(ct, "ensure_cloudflared", lambda: "/bin/cloudflared") + monkeypatch.setattr(ct, "CloudflareTunnel", _Stub) + assert ct.start_studio_tunnel(8080) is None + assert attempts == [None, "http2"] + assert ct._active_tunnel is None + + +def test_start_studio_tunnel_aborts_retry_on_concurrent_shutdown(monkeypatch): + # If a concurrent stop_studio_tunnel() clears _active_tunnel while we wait, + # the retry loop must NOT start a second (http2) tunnel: shutdown is already + # done, so nothing would ever stop it and it would be orphaned. + attempts = [] + + class _Stub: + def __init__( + self, + port, + binary, + protocol = None, + ): + self.url = None + attempts.append(protocol) + + def start(self): + self.url = "https://words.trycloudflare.com" # URL minted (saw_url True) + + def wait_for_ready(self, timeout): + # Simulate stop_studio_tunnel() landing during the wait. + with ct._active_lock: + ct._active_tunnel = None + return None # never registered + + def stop(self): + pass + + monkeypatch.setattr(ct, "ensure_cloudflared", lambda: "/bin/cloudflared") + monkeypatch.setattr(ct, "CloudflareTunnel", _Stub) + assert ct.start_studio_tunnel(8080) is None + assert attempts == [None] # no http2 retry -> no orphaned second tunnel + assert ct._active_tunnel is None + + # ── run.py source-level pins (AST / source, no heavy import) ───────── @@ -419,6 +696,13 @@ def test_argparse_cloudflare_default_true(): assert _argparse_default(_RUN_PY.read_text(), "--cloudflare") is True +def test_run_server_registers_tunnel_atexit_backstop(): + # An abnormal exit (exception after startup -> sys.exit) bypasses + # _graceful_shutdown; an atexit backstop must still stop the tunnel. + src = _RUN_PY.read_text() + assert "atexit.register(stop_studio_tunnel)" in src + + def test_run_server_gates_tunnel_on_wildcard(): # Guard against accidentally widening the trigger beyond 0.0.0.0. source = _RUN_PY.read_text()