Add optional Turnstile protection

Author: adewale

docs/lessons-learned.md

@@ -123,3 +123,4 @@ git diff --check
 - **Journey audit is graph audit plus outcome audit.** A journey section is healthy only when its examples form a prerequisite-respecting mental map and its declared outcomes are backed by cells on those examples. A section can have a beautiful figure and still fail if the support list is a catalog slice, if the `See also` graph isolates one example, or if the section caption describes a conceptual shift the examples do not actually make.
 - **A green total can still hide a weak criterion.** The journey-figure audit found every section figure above the project gate while 15 sections still reused a lesson paint function, which weakened the independence-from-lesson-figures dimension. Bespoke runtime, control-flow, iteration, types, and reliability section figures cleared that watchlist; keep tracking criterion-level weakness even when the aggregate score remains shippable.
 - **Deployment smoke belongs beside CI.** `scripts/smoke_deployment.py` checks rendered Worker pages, runtime-boundary pages, journey pages, prototype review pages, and representative Dynamic Worker POST runs for HTTP failures, exception markers, and stale edited-code output. Build success is not enough; the deployed Worker must render and execute edited examples.
+- **Turnstile should be secret-gated, not development-gated.** Protect edited-code POST runs only when `TURNSTILE_SECRET_KEY` is configured, render the widget only when `TURNSTILE_SITE_KEY` is present, and keep local/dev runs frictionless. If production smoke must POST through a protected endpoint, use a separate `PBE_SMOKE_BYPASS_SECRET` header so smoke remains a deployment check rather than a CAPTCHA solver.

scripts/lint_seo_cache.py

@@ -65,7 +65,7 @@ def assert_asset_manifest(failures: list[str]) -> None:
 def assert_worker_cache_policy(failures: list[str]) -> None:
     source = (ROOT / "src" / "main.py").read_text()
     required = [
-        "def html_cache_key_url(url: str) -> str:",
+        "def html_cache_key_url(url: str, turnstile_site_key: str = \"\") -> str:",
         "__html_v={HTML_CACHE_VERSION}",
         "caches.default.match(cache_key)",
         "caches.default.put(cache_key, response.clone())",

scripts/smoke_deployment.py

@@ -7,6 +7,7 @@
 from __future__ import annotations
 
 import argparse
+import os
 import sys
 import urllib.error
 import urllib.parse
@@ -39,15 +40,18 @@ def fetch(url: str) -> tuple[int, str]:
         return response.status, body
 
 
-def post_code(url: str, code: str) -> tuple[int, str]:
+def post_code(url: str, code: str, smoke_bypass_secret: str = "") -> tuple[int, str]:
     data = urllib.parse.urlencode({"code": code}).encode()
+    headers = {
+        "User-Agent": "pythonbyexample-smoke/1.0",
+        "Content-Type": "application/x-www-form-urlencoded",
+    }
+    if smoke_bypass_secret:
+        headers["x-pythonbyexample-smoke-secret"] = smoke_bypass_secret
     request = urllib.request.Request(
         url,
         data=data,
-        headers={
-            "User-Agent": "pythonbyexample-smoke/1.0",
-            "Content-Type": "application/x-www-form-urlencoded",
-        },
+        headers=headers,
         method="POST",
     )
     with urllib.request.urlopen(request, timeout=30) as response:
@@ -91,11 +95,13 @@ def main() -> int:
             failures.append(f"{url}: rendered exception marker {marker!r}")
         print(f"GET {status} {url}")
 
+    smoke_bypass_secret = os.environ.get("PBE_SMOKE_BYPASS_SECRET", "")
+
     if not args.skip_post:
         for slug, code, expected in POST_SMOKES:
             url = urljoin(base, f"examples/{slug}")
             try:
-                status, body = post_code(url, code)
+                status, body = post_code(url, code, smoke_bypass_secret)
             except urllib.error.HTTPError as exc:
                 failures.append(f"POST {url}: HTTP {exc.code}")
                 continue

src/app.py

@@ -748,7 +748,17 @@ def _render_cell(step):
     return f'<section class="lesson-step lp-cell"><div class="lp-prose">{prose_html}</div><div class="cell-code-stack"><div class="cell-source"><p class="cell-label">Source</p><pre><code class="language-python">{source}</code></pre></div><div class="cell-output"><p class="cell-label">Output</p><pre><code>{html.escape(step["output"])}</code></pre></div></div></section>'
 
 
-def render_example_page(example, output=None, code=None, execution_time_ms=None):
+def _turnstile_widget(site_key: str | None) -> str:
+    if not site_key:
+        return ""
+    escaped = html.escape(site_key)
+    return (
+        '<script src="https://challenges.cloudflare.com/turnstile/v0/api.js" async defer></script>'
+        f'<div class="cf-turnstile" data-sitekey="{escaped}"></div>'
+    )
+
+
+def render_example_page(example, output=None, code=None, execution_time_ms=None, turnstile_site_key=None):
     notes = [render_inline(note) for note in example.get("notes", [])]
     walkthrough = _walkthrough_cells(example)
     shown_output = output if output is not None else example.get("expected_output", "Run this example to see output here.")
@@ -796,6 +806,7 @@ def render_example_page(example, output=None, code=None, execution_time_ms=None)
             "SLUG": html.escape(example["slug"]),
             "EDITOR_ROWS": str(max(18, editable_code.count("\n") + 2)),
             "EDITABLE_CODE": html.escape(editable_code),
+            "TURNSTILE_WIDGET": _turnstile_widget(turnstile_site_key),
             "OUTPUT_PLACEHOLDER": " data-output-placeholder" if output is None else "",
             "OUTPUT_HEADING": html.escape(output_heading),
             "SHOWN_OUTPUT": html.escape(shown_output),
@@ -813,7 +824,7 @@ def render_example_page(example, output=None, code=None, execution_time_ms=None)
     )
 
 
-def route(url: str, method: str = "GET") -> AppResponse:
+def route(url: str, method: str = "GET", turnstile_site_key: str | None = None) -> AppResponse:
     without_scheme = url.split("://", 1)[-1]
     path_part = without_scheme.split("/", 1)[1] if "/" in without_scheme else ""
     path = ("/" + path_part.split("?", 1)[0]).rstrip("/") or "/"
@@ -850,6 +861,7 @@ def route(url: str, method: str = "GET") -> AppResponse:
             body = f'<h1>Example not found</h1><p class="meta">Try one of these nearby examples.</p><h2>Recommended examples</h2><ul>{recommendations}</ul>'
             return AppResponse(_layout("Not Found", body), status=404)
         return AppResponse(
-            render_example_page(example), headers={"Content-Type": "text/html; charset=utf-8"}
+            render_example_page(example, turnstile_site_key=turnstile_site_key),
+            headers={"Content-Type": "text/html; charset=utf-8"},
         )
     return AppResponse(_layout("Not Found", "<h1>Not found</h1>"), status=404)

src/asset_manifest.py

@@ -1,3 +1,3 @@
 # Generated by scripts/fingerprint_assets.py. Do not edit by hand.
 ASSET_PATHS = {'SITE_CSS': '/site.57a55415849b.css', 'SYNTAX_JS': '/syntax-highlight.3b6c7f730d46.js', 'EDITOR_JS': '/editor.a4a7766e1b9b.js'}
-HTML_CACHE_VERSION = '934a3a7aa7f5'
+HTML_CACHE_VERSION = '05479c782a32'

src/main.py

@@ -1,6 +1,7 @@
 import hashlib
+import json
 import time
-from urllib.parse import parse_qs, urlparse
+from urllib.parse import parse_qs, urlencode, urlparse
 
 from fastapi import FastAPI, Request
 from fastapi.responses import HTMLResponse, Response
@@ -12,8 +13,11 @@
 
 import worker_asgi_bridge as asgi
 
+TURNSTILE_VERIFY_URL = "https://challenges.cloudflare.com/turnstile/v0/siteverify"
+SMOKE_BYPASS_HEADER = "x-pythonbyexample-smoke-secret"
+
 try:
-    from js import Object, Request as JsRequest, caches
+    from js import Object, Request as JsRequest, caches, fetch as js_fetch
     from pyodide.ffi import create_once_callable, jsnull, to_js
 except ImportError:  # Allows editor tooling outside Workers.
     Object = None
@@ -22,6 +26,7 @@
     create_once_callable = None
     to_js = None
     caches = None
+    js_fetch = None
 
 app = FastAPI(title="Python By Example")
 _CURRENT_WORKER_REQUEST = None
@@ -48,9 +53,13 @@ def should_cache_get_url(url: str) -> bool:
     return not path.startswith("/layout-options/")
 
 
-def html_cache_key_url(url: str) -> str:
+def html_cache_key_url(url: str, turnstile_site_key: str = "") -> str:
     separator = "&" if "?" in url else "?"
-    return f"{url}{separator}__html_v={HTML_CACHE_VERSION}"
+    turnstile_fragment = ""
+    if turnstile_site_key:
+        digest = hashlib.sha256(turnstile_site_key.encode("utf-8")).hexdigest()[:8]
+        turnstile_fragment = f"&__turnstile={digest}"
+    return f"{url}{separator}__html_v={HTML_CACHE_VERSION}{turnstile_fragment}"
 
 
 @app.get("/favicon.svg")
@@ -64,9 +73,19 @@ async def home(request: Request):
     return _html(response.body, response.status)
 
 
+def _env_text(request: Request, name: str) -> str:
+    env = request.scope.get("env")
+    value = getattr(env, name, "") if env is not None else ""
+    return value if isinstance(value, str) else ""
+
+
+def _turnstile_site_key(request: Request) -> str:
+    return _env_text(request, "TURNSTILE_SITE_KEY")
+
+
 @app.get("/examples/{slug}", response_class=HTMLResponse)
 async def example_page(slug: str, request: Request):
-    response = route(str(request.url), method="GET")
+    response = route(str(request.url), method="GET", turnstile_site_key=_turnstile_site_key(request))
     return _html(response.body, response.status)
 
 
@@ -76,11 +95,66 @@ async def run_example(slug: str, request: Request):
     if example is None:
         return _html("<h1>Example not found</h1>", 404)
     body = (await request.body()).decode("utf-8")
-    submitted = parse_qs(body).get("code", [example["code"]])[0]
+    form = parse_qs(body)
+    submitted = form.get("code", [example["code"]])[0]
+    turnstile_token = form.get("cf-turnstile-response", [""])[0]
+    ok, message = await _verify_turnstile(request, turnstile_token)
+    if not ok:
+        return _html(
+            render_example_page(
+                example,
+                output=message,
+                code=submitted,
+                turnstile_site_key=_turnstile_site_key(request),
+            )
+        )
     started = time.perf_counter()
     output = await _run_example(request, example["slug"], submitted)
     elapsed_ms = (time.perf_counter() - started) * 1000
-    return _html(render_example_page(example, output=output, code=submitted, execution_time_ms=elapsed_ms))
+    return _html(
+        render_example_page(
+            example,
+            output=output,
+            code=submitted,
+            execution_time_ms=elapsed_ms,
+            turnstile_site_key=_turnstile_site_key(request),
+        )
+    )
+
+
+async def _verify_turnstile(request: Request, token: str) -> tuple[bool, str]:
+    secret = _env_text(request, "TURNSTILE_SECRET_KEY")
+    if not secret:
+        return True, ""
+
+    bypass_secret = _env_text(request, "PBE_SMOKE_BYPASS_SECRET")
+    if bypass_secret and request.headers.get(SMOKE_BYPASS_HEADER) == bypass_secret:
+        return True, ""
+
+    if not token:
+        return False, "Turnstile verification is required before running edited code. Please retry."
+    if js_fetch is None or JsRequest is None:
+        return False, "Turnstile verification is unavailable outside the Cloudflare runtime."
+
+    payload = {"secret": secret, "response": token}
+    remote_ip = request.headers.get("CF-Connecting-IP")
+    if remote_ip:
+        payload["remoteip"] = remote_ip
+    verify_request = JsRequest.new(
+        TURNSTILE_VERIFY_URL,
+        _to_js_object(
+            {
+                "method": "POST",
+                "body": urlencode(payload),
+                "headers": {"Content-Type": "application/x-www-form-urlencoded"},
+            }
+        ),
+    )
+    response = await js_fetch(verify_request)
+    result = json.loads(await response.text())
+    if result.get("success") is True:
+        return True, ""
+    return False, "Turnstile verification failed. Please refresh the challenge and try again."
 
 
 async def _run_example(request: Request, slug, code):
@@ -134,7 +208,8 @@ async def fetch(self, request):
         # POST requests and are intentionally never cached.
         if getattr(request, "method", None) == "GET" and caches is not None:
             if should_cache_get_url(getattr(request, "url", "")):
-                cache_key = JsRequest.new(html_cache_key_url(getattr(request, "url", "")), _to_js_object({"method": "GET"}))
+                turnstile_site_key = getattr(self.env, "TURNSTILE_SITE_KEY", "")
+                cache_key = JsRequest.new(html_cache_key_url(getattr(request, "url", ""), turnstile_site_key), _to_js_object({"method": "GET"}))
                 cached = await caches.default.match(cache_key)
                 if cached:
                     return cached

src/templates/example.html

@@ -16,6 +16,7 @@ <h2>Run the complete example</h2>
       <form class="runner-panel runner-editor" method="post" action="/examples/__SLUG__">
         <h3>Example code</h3>
         <textarea name="code" id="code-editor" spellcheck="false" rows="__EDITOR_ROWS__">__EDITABLE_CODE__</textarea>
+        __TURNSTILE_WIDGET__
         <div class="playground-toolbar">
           <button class="button" type="submit">Run</button>
           <button class="tool-button" type="button" data-reset onclick="resetCode()">Reset</button>
@@ -52,6 +53,8 @@ <h3>Example code</h3>
     if (nextOutput) outputPanel.innerHTML = nextOutput.innerHTML;
   } catch (error) {
     outputPanel.querySelector('code').textContent = 'Run failed: ' + error.message;
+  } finally {
+    if (window.turnstile) turnstile.reset();
   }
 });
 const hash = new URL(window.location.href).hash;

tests/test_app.py

@@ -305,6 +305,26 @@ def test_ui_polish_principles_are_applied(self):
         self.assertNotIn('<article class="card"', home)
         self.assertIn('class="example-shell"', html)
 
+    def test_turnstile_widget_is_optional_and_form_bound(self):
+        html = render_example_page(get_example("hello-world"))
+        self.assertNotIn("cf-turnstile", html)
+        self.assertNotIn("challenges.cloudflare.com/turnstile", html)
+
+        protected = render_example_page(get_example("hello-world"), turnstile_site_key="site-key-123")
+        self.assertIn("https://challenges.cloudflare.com/turnstile/v0/api.js", protected)
+        self.assertIn('class="cf-turnstile"', protected)
+        self.assertIn('data-sitekey="site-key-123"', protected)
+        self.assertIn("window.turnstile", protected)
+        self.assertIn("turnstile.reset", protected)
+
+    def test_turnstile_verification_is_secret_gated_in_worker(self):
+        main_source = (ROOT / "src" / "main.py").read_text()
+        self.assertIn("TURNSTILE_SECRET_KEY", main_source)
+        self.assertIn("cf-turnstile-response", main_source)
+        self.assertIn("/turnstile/v0/siteverify", main_source)
+        self.assertIn("PBE_SMOKE_BYPASS_SECRET", main_source)
+        self.assertIn("x-pythonbyexample-smoke-secret", main_source)
+
     def test_cf_workers_design_system_and_playground_lessons(self):
         html = render_example_page(get_example("hello-world"), output="hello world\n")
         css = (ROOT / "public" / "site.css").read_text()