From b54841a87bd61ae89855bdd36a80121ee2da5f88 Mon Sep 17 00:00:00 2001 From: Adam Wright Date: Sat, 3 Oct 2026 14:34:09 +0000 Subject: [PATCH] Gate the chat's websocket, and make a captcha pass expire From the review of the public surface (area 1b). Reproduced first, then fixed, then checked in a real browser through Turnstile's test keys. - The captcha gate was HTTP middleware, which never sees a websocket. A client opening the chat's socket.io websocket directly got the whole chat with no captcha. An ASGI middleware now refuses ungated websocket handshakes (close 1008, a 403 to the client). - Guests' only per-person limit was per session, and the client picks the session id. Guests are now also limited per captcha solve (100 messages per 3h, configurable), which a client cannot reset without solving again. - The cookie was value|HMAC(value) with no issue time: one solve was a pass for ever, for anyone. It now carries an issue time and a nonce, expires on the server after 12h, and renews while in use so an open tab keeps working. Old-format cookies are refused: readers solve once more. - verify_captcha called the blocking requests.post on the shared event loop for every anonymous POST, and a non-JSON reply was a 500. Now async httpx, failure treated as not verified, and the form is bounded (no files) -- it spooled unbounded uploads to the shared disk. - The Referer check returned 400 to readers arriving from any http page and stopped no one. Removed. - The API exemption is the prefix with its slash; /chat/guest/api-x was exempt too. tests/api/test_captcha_gate.py drives the real app over HTTP and websocket in its own process; with the websocket gate removed, its three websocket tests fail. Co-Authored-By: Claude Opus 5.5 --- bin/chat-chainlit.py | 33 +++++++ bin/chat-fastapi.py | 155 ++++++++++++++++++------------ src/util/captcha_cookie.py | 84 ++++++++++++++++ src/util/rate_limit.py | 6 +- tests/api/gate_probe.py | 113 ++++++++++++++++++++++ tests/api/test_captcha_gate.py | 71 ++++++++++++++ tests/util/test_captcha_cookie.py | 67 +++++++++++++ 7 files changed, 466 insertions(+), 63 deletions(-) create mode 100644 src/util/captcha_cookie.py create mode 100644 tests/api/gate_probe.py create mode 100644 tests/api/test_captcha_gate.py create mode 100644 tests/util/test_captcha_cookie.py diff --git a/bin/chat-chainlit.py b/bin/chat-chainlit.py index d53e0e3..37ea68b 100644 --- a/bin/chat-chainlit.py +++ b/bin/chat-chainlit.py @@ -4,6 +4,7 @@ # named with a hyphen, so it cannot be listed in [[tool.mypy.overrides]]. import asyncio import os +import time from collections.abc import Awaitable, Callable from pathlib import Path @@ -39,6 +40,7 @@ from handoff import seed from handoff.store import AnalysisHandoff, handoffs from handoff.window import acknowledgement, claimed_id +from util import captcha_cookie from util.chainlit_helpers import ( PrefixedS3StorageClient, is_feature_enabled, @@ -51,9 +53,11 @@ from util.config_yml.messages import TriggerEvent from util.logging import logging from util.orcid_provider import ORCIDOAuthProvider +from util.rate_limit import SlidingWindowLimiter, positive_int from util.secrets import ( SECRET_NAMES, get_db_uri, + get_secret, load_secrets_to_environ, mounted_secrets, ) @@ -522,6 +526,32 @@ async def answer_with_model(content: str, message_id: str) -> None: save_openai_metrics(message_id, openai_cb) +#: Guests' messages, limited per human check rather than per session. The +#: per-session quota is keyed on the session id, which the client chooses, so +#: a reconnect -- or a script -- started a fresh quota every time (review, 1b). +_per_solve = SlidingWindowLimiter( + limit=positive_int("CHAT_MESSAGES_PER_SOLVE", 100), + window=float(positive_int("CHAT_SOLVE_WINDOW_SECONDS", 3 * 60 * 60)), +) +PER_SOLVE_LIMITED = ( + "You've reached the limit on messages for now. Please try again later." +) + + +def solve_key() -> str: + """Which human check this connection passed, or failing that, where from.""" + environ = context.session.environ or {} + verdict = captcha_cookie.check( + captcha_cookie.from_cookie_header(environ.get("HTTP_COOKIE")), + get_secret("CLOUDFLARE_SECRET_KEY") or "", + time.time(), + ) + if verdict.ok: + return f"solve:{verdict.nonce}" + forwarded = str(environ.get("HTTP_X_FORWARDED_FOR", "")).split(",")[0].strip() + return f"ip:{forwarded or environ.get('REMOTE_ADDR', '')}" + + @cl.on_message async def main(message: cl.Message) -> None: # First, before any early return: a "yes" means the offer just made, so @@ -531,6 +561,9 @@ async def main(message: cl.Message) -> None: if await message_rate_limited(config): return + if cl.user_session.get("user") is None and not _per_solve.allow(solve_key()): + await cl.Message(content=PER_SOLVE_LIMITED).send() + return await static_messages(config, TriggerEvent.on_message) diff --git a/bin/chat-fastapi.py b/bin/chat-fastapi.py index 645c38f..627aeed 100644 --- a/bin/chat-fastapi.py +++ b/bin/chat-fastapi.py @@ -1,13 +1,12 @@ -import hashlib -import hmac import os import time from collections.abc import AsyncIterator, Awaitable, Callable from contextlib import asynccontextmanager from string import Template +from typing import Any from urllib.parse import urlsplit -import requests +import httpx from chainlit.utils import mount_chainlit from dotenv import load_dotenv from fastapi import FastAPI, Request, Response @@ -17,6 +16,7 @@ from api.analysis_summary import router as analysis_summary_router from api.answer import router as answer_router from api.handoff import router as handoff_router +from util import captcha_cookie from util.caller_token import load_verifying_key from util.captcha_scope import is_captcha_exempt from util.embedding_environment import EmbeddingEnvironment @@ -117,26 +117,67 @@ async def lifespan(_app: FastAPI) -> AsyncIterator[None]: HEADER_DONT_CACHE = {"Cache-Control": "no-store"} -def make_signature(value: str) -> str: - if CLOUDFLARE_SECRET_KEY is None: - raise ValueError("CLOUDFLARE_SECRET_KEY is not set") - return hmac.new( - CLOUDFLARE_SECRET_KEY.encode(), value.encode(), hashlib.sha256 - ).hexdigest() +def _gated(path: str) -> bool: + """Whether the human check applies to this path.""" + return not is_captcha_exempt( + path, + chainlit_uri=CHAINLIT_URI, + # The value resolved through get_secret, not os.environ. get_secret + # prefers a mounted Docker secret file, so a deployment that mounts the + # key rather than exporting it used to land here with the env var unset + # and skip the captcha entirely -- switching off a protection the + # operator had configured, silently. + captcha_configured=bool(CLOUDFLARE_SECRET_KEY), + # The website API verifies its own caller, with a signed token rather + # than a captcha, so redirecting it to a captcha page would break it. This + # is a deliberate hole in an authentication boundary, which is why the + # rule was pinned by tests before it was widened. With the slash: the + # bare prefix exempted `/chat/guest/api-anything` too (review, 1b). + extra_prefixes=[f"{API_PREFIX}/"], + ) -def create_secure_cookie(value: str) -> str: - signature = make_signature(value) - return f"{value}|{signature}" +def _set_pass(response: Response) -> None: + response.set_cookie( + key=captcha_cookie.COOKIE_NAME, + value=captcha_cookie.mint(CLOUDFLARE_SECRET_KEY or "", time.time()), + max_age=captcha_cookie.MAX_AGE_SECONDS, + secure=True, # HTTPS only + httponly=True, # inaccessible to client side JS + ) -def verify_secure_cookie(cookie_value: str) -> bool: - try: - value, signature = cookie_value.split("|", 1) - expected_signature = make_signature(value) - return hmac.compare_digest(signature, expected_signature) - except Exception: - return False +class WebsocketGate: + """The human check, for websocket connections. + + `@app.middleware("http")` never sees a websocket. Browsers met the gate only + because socket.io starts on HTTP polling; a client opening the websocket + directly got the whole chat with no captcha, and -- choosing its own session + id -- no per-person limit either (review, area 1b, reproduced). + + Refused before the handshake completes, which the server sends as a 403. + """ + + def __init__(self, app: Any) -> None: + self.app = app + + async def __call__(self, scope: Any, receive: Any, send: Any) -> None: + if scope["type"] == "websocket" and _gated(scope.get("path", "")): + headers = dict(scope.get("headers") or []) + cookie = captcha_cookie.from_cookie_header( + headers.get(b"cookie", b"").decode("latin-1") + ) + verdict = captcha_cookie.check( + cookie, CLOUDFLARE_SECRET_KEY or "", time.time() + ) + if not verdict.ok: + await receive() # websocket.connect + await send({"type": "websocket.close", "code": 1008}) + return + await self.app(scope, receive, send) + + +app.add_middleware(WebsocketGate) @app.middleware("http") @@ -151,39 +192,27 @@ async def verify_captcha_middleware( if ".." not in clean_path: return RedirectResponse(url=f"{clean_path}/") - # Allow access to CAPTCHA pages and static files - if is_captcha_exempt( - path, - chainlit_uri=CHAINLIT_URI, - # The value resolved through get_secret, not os.environ. get_secret - # prefers a mounted Docker secret file, so a deployment that mounts the - # key rather than exporting it used to land here with the env var unset - # and skip the captcha entirely -- switching off a protection the - # operator had configured, silently. - captcha_configured=bool(CLOUDFLARE_SECRET_KEY), - # The answer endpoint verifies its own caller, with a signed token rather - # than a captcha, so redirecting it to a captcha page would break it. This - # is a deliberate hole in an authentication boundary, which is why the - # rule was pinned by tests before it was widened. - extra_prefixes=[API_PREFIX], - ): + if not _gated(path): return await call_next(request) - host = request.headers.get("referer") - if host and host.startswith("http:"): - error_html = ERROR_PAGE_TEMPLATE.substitute( - error_title="HTTPS is required for accessing this site", - ) - return Response(content=error_html, status_code=400, media_type="text/html") - - # Check if the user has completed the CAPTCHA verification - captcha_verified = request.cookies.get("captcha_verified") + # There was a check here that refused any request whose Referer was an + # http: page. The Referer is the page the reader came from, so it turned + # away readers following a link from any plain-http site, and stopped no + # one: a script omits the header (review, area 1b). TLS is Apache's job, + # and the cookie is Secure. - # If CAPTCHA is not verified, block access - if not captcha_verified or not verify_secure_cookie(captcha_verified): + verdict = captcha_cookie.check( + request.cookies.get(captcha_cookie.COOKIE_NAME), + CLOUDFLARE_SECRET_KEY or "", + time.time(), + ) + if not verdict.ok: return RedirectResponse(url=f"{CHAINLIT_URI}/verify_captcha_page") - return await call_next(request) + response = await call_next(request) + if verdict.renew: + _set_pass(response) + return response # Serve the CAPTCHA verification page (basic HTML form) @@ -227,7 +256,10 @@ async def captcha_page() -> Response: @app.post(f"{CHAINLIT_URI}/verify_captcha") async def verify_captcha(request: Request) -> Response: - form_data = await request.form() + # Bounded: anyone can reach this route, and the defaults put no cap on + # file parts, which were spooled to the disk the whole host shares + # (review, area 1b). The form has one field. + form_data = await request.form(max_files=0, max_fields=4, max_part_size=8192) cf_turnstile_response = form_data.get("cf-turnstile-response") if not isinstance(cf_turnstile_response, str): error_html = ERROR_PAGE_TEMPLATE.substitute( @@ -267,9 +299,18 @@ async def verify_captcha(request: Request) -> Response: "remoteip": client_ip, } - # Perform request to Cloudflare Turnstile verification endpoint - response = requests.post(url, data=data, timeout=10) - result = response.json() + # Asynchronous: the blocking `requests.post` held the event loop every + # session shares for a Cloudflare round trip on each anonymous POST, and + # an unreachable or non-JSON reply was an unhandled 500 (review, 1b). + try: + async with httpx.AsyncClient(timeout=10.0) as client: + reply = await client.post(url, data=data) + result = reply.json() + except (httpx.HTTPError, ValueError): + logging.warning("Turnstile siteverify failed; treating as not verified") + result = {} + if not isinstance(result, dict): + result = {} # If CAPTCHA validation fails, return an error if not result.get("success"): @@ -283,18 +324,12 @@ async def verify_captcha(request: Request) -> Response: media_type="text/html", ) - # Set a signed cookie to mark CAPTCHA as verified - cookie_value = create_secure_cookie(cf_turnstile_response) redirect_response = RedirectResponse( url=f"{CHAINLIT_URI}/", status_code=302, headers=HEADER_DONT_CACHE ) - redirect_response.set_cookie( - key="captcha_verified", - value=cookie_value, - max_age=3600, # Cookie expires in 1 hour - secure=True, # HTTPS only - httponly=True, # inaccessible to client side JS - ) + # A fresh nonce and issue time -- not the Turnstile token, which the old + # cookie carried and which named nothing a limit could count. + _set_pass(redirect_response) return redirect_response diff --git a/src/util/captcha_cookie.py b/src/util/captcha_cookie.py new file mode 100644 index 0000000..744e5be --- /dev/null +++ b/src/util/captcha_cookie.py @@ -0,0 +1,84 @@ +"""The cookie that says this browser passed the human check. + +Review of the public surface (area 1b, 2026-10-03) found the first version was +`value|HMAC(secret, value)` with no issue time. `max_age` is only a hint to the +browser, so one solved challenge was a pass for ever, for any client, and -- +since beta signs with production's Turnstile secret -- on production too. + +Now the signed value carries its issue time and a random nonce: + +- **It expires on the server.** A cookie older than `MAX_AGE_SECONDS` fails, + whatever the browser was told. +- **It renews while in use.** One older than `RENEW_AFTER_SECONDS` is re-issued + on the next HTTP response, so an open tab keeps working past the first hour + instead of its uploads and buttons starting to fail. +- **The nonce names one solve.** Rate limits key on it: unlike Chainlit's + session id, the client cannot mint a new one without solving another + challenge. + +Pure functions over the secret and the clock, so the rules are tested here; the +gate in `bin/chat-fastapi.py` is wiring. +""" + +import hashlib +import hmac +import secrets +from dataclasses import dataclass +from http.cookies import CookieError, SimpleCookie + +COOKIE_NAME = "captcha_verified" +VERSION = "v2" +MAX_AGE_SECONDS = 12 * 60 * 60 +RENEW_AFTER_SECONDS = 30 * 60 +#: A clock a little behind ours must not make a fresh cookie look from the +#: future and fail. +FUTURE_SKEW_SECONDS = 60 + + +@dataclass(frozen=True) +class Check: + ok: bool + #: Which solve this is, for rate limiting. Empty unless `ok`. + nonce: str = "" + #: Old enough that the response should carry a fresh cookie. + renew: bool = False + + +def _sign(secret: str, value: str) -> str: + return hmac.new(secret.encode(), value.encode(), hashlib.sha256).hexdigest() + + +def mint(secret: str, now: float) -> str: + """A fresh cookie value: version, issue time, nonce, signature.""" + value = f"{VERSION}.{int(now)}.{secrets.token_urlsafe(16)}" + return f"{value}|{_sign(secret, value)}" + + +def check(cookie_value: str | None, secret: str, now: float) -> Check: + """Whether a cookie value is a current pass. Never raises.""" + if not cookie_value or not secret: + return Check(ok=False) + value, _, signature = cookie_value.partition("|") + if not hmac.compare_digest(signature, _sign(secret, value)): + return Check(ok=False) + parts = value.split(".") + if len(parts) != 3 or parts[0] != VERSION or not parts[1].isdigit(): + # Signed but in the old, timeless format: a pass that never expired. + return Check(ok=False) + age = now - int(parts[1]) + if age > MAX_AGE_SECONDS or age < -FUTURE_SKEW_SECONDS: + return Check(ok=False) + return Check(ok=True, nonce=parts[2], renew=age > RENEW_AFTER_SECONDS) + + +def from_cookie_header(header: str | None) -> str | None: + """This cookie's value from a raw `Cookie:` header, or None.""" + if not header: + return None + jar: SimpleCookie = SimpleCookie() + try: + jar.load(header) + except CookieError: + return None + morsel = jar.get(COOKIE_NAME) + return morsel.value if morsel else None diff --git a/src/util/rate_limit.py b/src/util/rate_limit.py index 0b0b321..2aa1c93 100644 --- a/src/util/rate_limit.py +++ b/src/util/rate_limit.py @@ -15,7 +15,7 @@ from collections import deque -def _positive_int(name: str, default: int) -> int: +def positive_int(name: str, default: int) -> int: """Configuration that is absent, empty or nonsense falls back to the default.""" raw = os.getenv(name, "") if not raw.strip(): @@ -98,6 +98,6 @@ def limiter_from_env() -> SlidingWindowLimiter: than anyone reads -- and it still caps a leaked token at 180 an hour. """ return SlidingWindowLimiter( - limit=_positive_int("ANSWER_RATE_LIMIT", 30), - window=float(_positive_int("ANSWER_RATE_WINDOW_SECONDS", 600)), + limit=positive_int("ANSWER_RATE_LIMIT", 30), + window=float(positive_int("ANSWER_RATE_WINDOW_SECONDS", 600)), ) diff --git a/tests/api/gate_probe.py b/tests/api/gate_probe.py new file mode 100644 index 0000000..a8701a9 --- /dev/null +++ b/tests/api/gate_probe.py @@ -0,0 +1,113 @@ +"""Drive the real captcha gate, in its own process, and print what happened. + +Run by `test_captcha_gate.py`. A separate process because importing +`bin/chat-fastapi.py` mounts Chainlit, whose configuration is global and would +leak into the rest of the suite. The lifespan (which builds the graph) is never +entered, so this takes seconds, not minutes. +""" + +import hashlib +import hmac +import importlib.util +import json +import os +import sys +import time +from pathlib import Path + +REPO = Path(__file__).resolve().parents[2] +SECRET = "test-secret-not-a-real-key" # noqa: S105 + +os.environ.update( + { + "CLOUDFLARE_SECRET_KEY": SECRET, + "CLOUDFLARE_SITE_KEY": "test-site-key", + "CHAINLIT_URI": "/chat/guest", + "CHAINLIT_URL": "https://testserver", + "OPENAI_API_KEY": "sk-test", + "LOG_LEVEL": "error", + } +) +sys.path.insert(0, str(REPO / "src")) +os.chdir(REPO) + +spec = importlib.util.spec_from_file_location( + "chat_fastapi", REPO / "bin/chat-fastapi.py" +) +assert spec is not None +assert spec.loader is not None +module = importlib.util.module_from_spec(spec) +spec.loader.exec_module(module) + +from fastapi.testclient import TestClient # noqa: E402 +from starlette.websockets import WebSocketDisconnect # noqa: E402 + +from util import captcha_cookie # noqa: E402 + +client = TestClient(module.app, base_url="https://testserver") +WS = "/chat/guest/ws/socket.io/?EIO=4&transport=websocket" +results: dict[str, object] = {} + + +def fresh() -> str: + return captcha_cookie.mint(SECRET, time.time()) + + +def aged(seconds: float) -> str: + return captcha_cookie.mint(SECRET, time.time() - seconds) + + +def old_format() -> str: + value = "a-turnstile-token-from-2024" + return f"{value}|{hmac.new(SECRET.encode(), value.encode(), hashlib.sha256).hexdigest()}" + + +def http(path: str, cookie: str | None = None, **headers: str) -> tuple[int, str, bool]: + client.cookies.clear() + if cookie: + client.cookies.set(captcha_cookie.COOKIE_NAME, cookie) + response = client.get(path, headers=headers, follow_redirects=False) + renewed = captcha_cookie.COOKIE_NAME in response.headers.get("set-cookie", "") + return response.status_code, response.headers.get("location", ""), renewed + + +def ws(cookie: str | None) -> str: + client.cookies.clear() + # engine.io refuses a TestClient websocket without these. + headers = {"upgrade": "websocket", "connection": "Upgrade"} + if cookie: + headers["cookie"] = f"{captcha_cookie.COOKIE_NAME}={cookie}" + try: + with client.websocket_connect(WS, headers=headers) as socket: + return "open:" + socket.receive_text()[:1] + except WebSocketDisconnect as closed: + return f"refused:{closed.code}" + + +results["http_no_cookie"] = http("/chat/guest/") +results["http_fresh"] = http("/chat/guest/", fresh()) +results["http_old_format"] = http("/chat/guest/", old_format()) +results["http_expired"] = http( + "/chat/guest/", aged(captcha_cookie.MAX_AGE_SECONDS + 60) +) +results["http_renewed"] = http( + "/chat/guest/", aged(captcha_cookie.RENEW_AFTER_SECONDS + 60) +) +results["http_from_plain_http_page"] = http( + "/chat/guest/", fresh(), referer="http://example.org/blog" +) +results["http_api_lookalike"] = http("/chat/guest/api-anything") +results["ws_no_cookie"] = ws(None) +results["ws_old_format"] = ws(old_format()) +results["ws_expired"] = ws(aged(captcha_cookie.MAX_AGE_SECONDS + 60)) +results["ws_fresh"] = ws(fresh()) + +client.cookies.clear() +upload = client.post( + "/chat/guest/verify_captcha", + files={"big": ("big.bin", b"x" * 200_000)}, + follow_redirects=False, +) +results["verify_with_a_file"] = upload.status_code + +print(json.dumps(results)) diff --git a/tests/api/test_captcha_gate.py b/tests/api/test_captcha_gate.py new file mode 100644 index 0000000..9ae5398 --- /dev/null +++ b/tests/api/test_captcha_gate.py @@ -0,0 +1,71 @@ +"""The human check, driven through the real app over HTTP and websocket. + +Review of the public surface (area 1b, 2026-10-03): the gate was HTTP +middleware, which never sees a websocket, so a client opening the chat's +socket directly got the whole chat with no captcha. Nothing caught it because +no test drove the real gate -- the existing ones grep source or test a copied +snippet. These run `gate_probe.py`, which imports `bin/chat-fastapi.py` in its +own process (Chainlit's configuration is global) and reports what happened. +""" + +import json +import subprocess +import sys +from pathlib import Path +from typing import Any + +import pytest + +PROBE = Path(__file__).with_name("gate_probe.py") + + +@pytest.fixture(scope="module") +def gate() -> dict[str, Any]: + done = subprocess.run( # noqa: S603 + [sys.executable, str(PROBE)], + capture_output=True, + text=True, + timeout=180, + check=False, + ) + assert done.returncode == 0, done.stderr[-2000:] + return dict(json.loads(done.stdout.strip().splitlines()[-1])) + + +GATED = [307, "/chat/guest/verify_captcha_page", False] + + +@pytest.mark.parametrize("case", ["ws_no_cookie", "ws_old_format", "ws_expired"]) +def test_the_websocket_is_gated(gate: dict[str, Any], case: str) -> None: + assert gate[case] == "refused:1008" + + +def test_a_passed_check_opens_the_websocket(gate: dict[str, Any]) -> None: + # The control: without it, a gate that refused everything would pass. + assert gate["ws_fresh"] == "open:0" + + +@pytest.mark.parametrize( + "case", ["http_no_cookie", "http_old_format", "http_expired", "http_api_lookalike"] +) +def test_http_without_a_current_pass_is_sent_to_the_check( + gate: dict[str, Any], case: str +) -> None: + assert gate[case] == GATED + + +def test_a_current_pass_is_let_through_and_renewed_when_ageing( + gate: dict[str, Any], +) -> None: + assert gate["http_fresh"] == [200, "", False] + assert gate["http_renewed"] == [200, "", True] + + +def test_a_reader_from_a_plain_http_page_is_let_in(gate: dict[str, Any]) -> None: + # The Referer check returned 400 to anyone following a link from an http + # page, and stopped no one. + assert gate["http_from_plain_http_page"] == [200, "", False] + + +def test_the_check_form_takes_no_files(gate: dict[str, Any]) -> None: + assert gate["verify_with_a_file"] == 400 diff --git a/tests/util/test_captcha_cookie.py b/tests/util/test_captcha_cookie.py new file mode 100644 index 0000000..254b07e --- /dev/null +++ b/tests/util/test_captcha_cookie.py @@ -0,0 +1,67 @@ +"""The captcha pass: expires on the server, renews in use, names one solve.""" + +import hashlib +import hmac + +from util import captcha_cookie as cookie + +SECRET = "test-secret" # noqa: S105 +NOW = 1_800_000_000.0 + + +def test_a_fresh_pass_checks_out_and_names_its_solve() -> None: + verdict = cookie.check(cookie.mint(SECRET, NOW), SECRET, NOW) + assert verdict.ok + assert len(verdict.nonce) >= 16 + assert not verdict.renew + + +def test_two_solves_have_different_nonces() -> None: + a = cookie.check(cookie.mint(SECRET, NOW), SECRET, NOW) + b = cookie.check(cookie.mint(SECRET, NOW), SECRET, NOW) + assert a.nonce != b.nonce + + +def test_it_expires_on_the_server_whatever_the_browser_was_told() -> None: + minted = cookie.mint(SECRET, NOW) + assert cookie.check(minted, SECRET, NOW + cookie.MAX_AGE_SECONDS - 1).ok + assert not cookie.check(minted, SECRET, NOW + cookie.MAX_AGE_SECONDS + 1).ok + + +def test_it_asks_to_be_renewed_while_in_use() -> None: + minted = cookie.mint(SECRET, NOW) + assert cookie.check(minted, SECRET, NOW + cookie.RENEW_AFTER_SECONDS + 1).renew + + +def test_the_old_timeless_format_is_refused_even_when_signed() -> None: + # value|HMAC(value): a pass that never expired (review, area 1b). + value = "turnstile-token" + signed = f"{value}|{hmac.new(SECRET.encode(), value.encode(), hashlib.sha256).hexdigest()}" + assert not cookie.check(signed, SECRET, NOW).ok + + +def test_tampering_and_other_secrets_fail() -> None: + minted = cookie.mint(SECRET, NOW) + value, _, signature = minted.partition("|") + later = value.replace(value.split(".")[1], str(int(NOW) + 10_000)) + assert not cookie.check(f"{later}|{signature}", SECRET, NOW + 10_000).ok + assert not cookie.check(minted, "another-secret", NOW).ok + assert not cookie.check(minted, "", NOW).ok + + +def test_a_pass_from_far_in_the_future_fails() -> None: + minted = cookie.mint(SECRET, NOW + 3600) + assert not cookie.check(minted, SECRET, NOW).ok + assert cookie.check(cookie.mint(SECRET, NOW + 30), SECRET, NOW).ok + + +def test_garbage_never_raises() -> None: + for junk in (None, "", "|", "a|b|c", "v2.x.y|sig", "\x00"): + assert not cookie.check(junk, SECRET, NOW).ok + + +def test_reading_it_from_a_cookie_header() -> None: + header = f"other=1; {cookie.COOKIE_NAME}=v2.1.abc|def; x=y" + assert cookie.from_cookie_header(header) == "v2.1.abc|def" + assert cookie.from_cookie_header(None) is None + assert cookie.from_cookie_header("other=1") is None