From 543652d0545c229e7872e51caa29861fe0d96f69 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 2 Nov 2026 14:02:41 +0000 Subject: [PATCH 1/2] feat(journeys): journey file format and prescale journeys verify --- CHANGELOG.md | 13 + README.md | 39 +++ cli/src/prescale_cli/commands/journeys.py | 164 +++++++++++ cli/src/prescale_cli/journeys/__init__.py | 6 + cli/src/prescale_cli/journeys/model.py | 332 ++++++++++++++++++++++ cli/src/prescale_cli/journeys/replay.py | 133 +++++++++ cli/src/prescale_cli/main.py | 3 + cli/tests/test_journeys.py | 260 +++++++++++++++++ examples/README.md | 9 + examples/fragile-shop.py | 71 ++++- examples/journeys/browse.yaml | 12 + examples/journeys/checkout.yaml | 25 ++ 12 files changed, 1064 insertions(+), 3 deletions(-) create mode 100644 cli/src/prescale_cli/commands/journeys.py create mode 100644 cli/src/prescale_cli/journeys/__init__.py create mode 100644 cli/src/prescale_cli/journeys/model.py create mode 100644 cli/src/prescale_cli/journeys/replay.py create mode 100644 cli/tests/test_journeys.py create mode 100644 examples/journeys/browse.yaml create mode 100644 examples/journeys/checkout.yaml diff --git a/CHANGELOG.md b/CHANGELOG.md index 0240859..e2909bb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,19 @@ ## Unreleased +### Added +- **Journeys: your users' real paths, saved in your repo.** A journey file + (`prescale/journeys/.yaml`) lists the requests a user's path makes (log in, add to + cart, place an order), carrying tokens and IDs from one step to the next. + `prescale journeys` lists them; `prescale journeys verify ` replays each once, as one + user, and names the step that no longer answers as expected (the journey is stale, or the + app is broken). It exits non-zero, so it works as a CI check. + - Secrets come from environment variables (`{{env.NAME}}`); a literal password, token or + API key in a journey file is refused. + - Journeys that change data only run against local hosts or hosts you allow with + `--allow-writes`; journeys that send messages or delete data are never replayed. + - The example shop has a JSON API and two example journeys in `examples/journeys/`. + ### Fixed - **`prescale mcp` works on a fresh install again.** `pip install 'prescale[mcp]'` picked up the new mcp 2.x, which renamed the server class PreScale uses, so the MCP server wouldn't start and told diff --git a/README.md b/README.md index 178406b..552c8a1 100644 --- a/README.md +++ b/README.md @@ -94,6 +94,7 @@ From source: `git clone https://github.com/pyjeebz/PreScale.git && pip install . | [`run `](#run--what-breaks-first) | Ramp traffic; report what breaks first and at what load | | [`investigate `](#investigate--and-why) | …and *why* it breaks, plus how to fix it (`--app` to start your app first: [replica mode](#replica-mode--test-your-app-not-your-host)) | | [`audit `](#audit--hygiene-no-load) | Load-free HTTP hygiene check (compression, caching, CDN, HTTP/2) | +| [`journeys`](#journeys--your-users-real-paths) | List your journey files; `journeys verify ` checks they still replay | | [`compare`](#ci--gate-on-capacity) | Capacity diff of two runs; regression gating for CI | | [`profiles`](#frame-it-as-a-launch) | Launch scenarios for `--profile` | | [`history` / `show`](#saved-runs-and-the-json-contract) | List and re-open saved runs | @@ -166,6 +167,44 @@ Try this → Add a pooler (e.g. pgbouncer) and check the pool checkout timeout. ``` +### `journeys` — your users' real paths + +A journey is a path a user takes through your app (log in, add to cart, place an order), saved as the requests it makes. Journeys live in your repo as `prescale/journeys/.yaml`: + +```yaml +version: 1 +name: checkout +type: checkout +steps: + - name: log in + request: + method: POST + path: /api/login + json: {email: "{{unique.email}}", password: "{{env.SHOP_PASSWORD}}"} + expect: {status: 200} + extract: {token: body.token} + - name: place order + request: + method: POST + path: /api/orders + headers: {Authorization: "Bearer {{token}}"} + expect: {status: 201} +``` + +```bash +prescale journeys # list them +prescale journeys verify http://localhost:3000 # replay each once, as one user +``` + +`verify` stops at the first step that doesn't answer as expected and calls the journey **stale**: the app changed, or it's broken. It exits non-zero, so it works as a CI check. + +- `{{name}}` is a value an earlier step pulled out with `extract` (`body.` or `header.`); `{{unique.email}}` and `{{unique.id}}` are fresh each run; `{{env.NAME}}` reads an environment variable. +- **Secrets never go in the file.** A literal password, token, API key or `Authorization` value is refused; use `{{env.NAME}}`. +- Paths must start with `/`: a journey only calls the app it's run against. +- Journeys that change data (POST, PUT, PATCH, DELETE) run against local hosts, or a host you name with `--allow-writes staging.myapp.com`. Types `side_effect` (emails, SMS, invites) and `destructive` are never replayed. + +Try it on the example shop: `python examples/fragile-shop.py 8400`, then `SHOP_PASSWORD=demo prescale journeys verify http://localhost:8400 --dir examples/journeys`. + ### `audit` — hygiene, no load A fast, load-free check of the HTTP-level footguns that decide how you scale — compression, static-asset caching, CDN, HTTP version, cookies on assets. Cheap enough to run on every commit. diff --git a/cli/src/prescale_cli/commands/journeys.py b/cli/src/prescale_cli/commands/journeys.py new file mode 100644 index 0000000..626b39a --- /dev/null +++ b/cli/src/prescale_cli/commands/journeys.py @@ -0,0 +1,164 @@ +"""`prescale journeys` - list the app's journeys, and check they still replay. + +Journeys live in the app's repo as `prescale/journeys/*.yaml`. `verify` replays +each one once, as a single user: a step that fails there means the journey is +stale (the app changed) or the app is broken, never that it broke under load. +""" + +from __future__ import annotations + +import asyncio +import os +from pathlib import Path +from urllib.parse import urlparse + +import click +from rich.console import Console +from rich.markup import escape +from rich.table import Table + +from prescale_cli.commands.upload import header_option +from prescale_cli.journeys.model import DEFAULT_DIR, Journey, load_dir +from prescale_cli.journeys.replay import JourneyCheck, check + +console = Console() + +_LOCAL_HOSTS = {"localhost", "127.0.0.1", "0.0.0.0", "::1"} + +dir_option = click.option( + "--dir", "directory", type=click.Path(file_okay=False, path_type=Path), + default=DEFAULT_DIR, show_default=True, help="Where the journey files are.") + + +@click.group(invoke_without_command=True) +@dir_option +@click.pass_context +def journeys(ctx: click.Context, directory: Path) -> None: + """List the app's journeys (prescale/journeys/*.yaml). + + \b + Check they still replay against your app: + prescale journeys verify http://localhost:3000 + """ + if ctx.invoked_subcommand is not None: + return + found = _load_or_exit(directory) + table = Table(title=f"Journeys in {directory}") + for col in ("Name", "Type", "Weight", "Steps", "Changes data", "File"): + table.add_column(col) + for j in found: + table.add_row(j.name, j.type, f"{j.weight:g}", str(len(j.steps)), + "yes" if j.writes else "no", str(j.source)) + console.print(table) + + +@journeys.command() +@click.argument("url") +@dir_option +@click.option("--journey", "only", multiple=True, + help="Check only this journey (repeatable).") +@click.option("--allow-writes", "allow_writes", multiple=True, metavar="HOST", + help="Let journeys that change data (POST, PUT, PATCH, DELETE) run against " + "this non-local host, e.g. staging.myapp.com (repeatable).") +@click.option("--timeout", default=10.0, type=float, help="Per-request timeout in seconds.") +@header_option +def verify(url: str, directory: Path, only: tuple[str, ...], allow_writes: tuple[str, ...], + timeout: float, headers: dict[str, str]) -> None: + """Replay each journey once, as one user, and report any step that fails. + + Exits non-zero if a journey is stale or the app couldn't be reached. + + \b + prescale journeys verify http://localhost:3000 + prescale journeys verify https://staging.myapp.com --allow-writes staging.myapp.com + """ + parsed = urlparse(url) + if parsed.scheme not in ("http", "https") or not parsed.netloc: + console.print(f"[red]Error:[/red] '{url}' is not a URL " + "(expected e.g. http://localhost:3000).") + raise SystemExit(2) + found = _load_or_exit(directory) + if only: + missing = sorted(set(only) - {j.name for j in found}) + if missing: + console.print(f"[red]Error:[/red] no journey named {', '.join(missing)} " + f"in {directory}.") + raise SystemExit(2) + found = [j for j in found if j.name in only] + + host = (parsed.hostname or "").lower() + _refuse_unallowed_writes(found, host, {h.lower() for h in allow_writes}) + unset = sorted({name for j in found for name in j.env_names()} - set(os.environ)) + if unset: + console.print(f"[red]Error:[/red] the journeys read environment variable(s) that " + f"aren't set: {', '.join(unset)}.") + raise SystemExit(2) + + base = url if url.endswith("/") else url + "/" + results = asyncio.run(_check_all(found, base, timeout=timeout, headers=headers)) + for result in results: + _print_check(result) + + broken = [r for r in results if not r.ok] + if broken: + stale = [r.journey.name for r in broken if r.stale] + if stale: + console.print(f"\n[yellow]{len(stale)} stale:[/yellow] {', '.join(stale)}. The app " + "no longer answers the way these journeys expect. If the change was " + "on purpose, update the journey files; if not, it's a bug.") + unreached = [r.journey.name for r in broken if not r.stale] + if unreached: + console.print(f"\n[red]{len(unreached)} couldn't run:[/red] {', '.join(unreached)}.") + raise SystemExit(1) + console.print(f"\n[green]✓ All {len(results)} journeys replay.[/green]") + + +def _load_or_exit(directory: Path) -> list[Journey]: + found, errors = load_dir(directory) + for error in errors: + console.print(f"[red]Error:[/red] {escape(error)}", highlight=False) + if errors: + raise SystemExit(2) + if not found: + console.print(f"No journeys in {directory}. Add one as {directory}/.yaml.") + raise SystemExit(1) + return found + + +def _refuse_unallowed_writes(found: list[Journey], host: str, allowed: set[str]) -> None: + """Journeys that change data run on local hosts, or on hosts named with + --allow-writes. Anywhere else they could change real data.""" + if host in _LOCAL_HOSTS or host in allowed: + return + writers = [j.name for j in found if j.writes] + if writers: + console.print(f"[red]Error:[/red] {', '.join(writers)} change data (POST, PUT, PATCH " + f"or DELETE). Run them against {host} only if it's a staging or test " + f"environment, with --allow-writes {host}.") + raise SystemExit(2) + + +async def _check_all(found: list[Journey], base: str, *, timeout: float, + headers: dict[str, str]) -> list[JourneyCheck]: + return [await check(j, base, timeout=timeout, headers=headers) for j in found] + + +def _print_check(result: JourneyCheck) -> None: + journey = result.journey + mark = "[green]✓[/green]" if result.ok else ("[yellow]stale[/yellow]" if result.stale + else "[red]✗[/red]") + console.print(f"\n{mark} [bold]{journey.name}[/bold] [dim]({journey.type}, " + f"{len(journey.steps)} steps)[/dim]") + for i, step in enumerate(journey.steps, 1): + done = result.steps[i - 1] if i <= len(result.steps) else None + if done is None: + console.print(f" [dim]{i}. {step.name} {step.label} (not run)[/dim]") + elif done.ok: + console.print(f" {i}. {step.name} [dim]{step.label}[/dim] " + f"{done.status} {done.latency * 1000:.0f}ms") + else: + console.print(f" {i}. {step.name} [dim]{step.label}[/dim] " + f"[red]{escape(done.problem or '')}[/red]", highlight=False) + if done.status is not None and done.status >= 500: + console.print(" [dim]A 5xx with no load is usually a bug in the app, " + "not a stale journey.[/dim]") diff --git a/cli/src/prescale_cli/journeys/__init__.py b/cli/src/prescale_cli/journeys/__init__.py new file mode 100644 index 0000000..3dab79d --- /dev/null +++ b/cli/src/prescale_cli/journeys/__init__.py @@ -0,0 +1,6 @@ +"""Journeys: a user's path through the app (log in, search, add to cart, order), +saved as the HTTP requests it makes, so PreScale can replay it. + +A journey lives in the app's own repo as `prescale/journeys/.yaml`. +`model` loads and checks those files; `replay` runs a journey's steps. +""" diff --git a/cli/src/prescale_cli/journeys/model.py b/cli/src/prescale_cli/journeys/model.py new file mode 100644 index 0000000..1ca708b --- /dev/null +++ b/cli/src/prescale_cli/journeys/model.py @@ -0,0 +1,332 @@ +"""The journey file format (version 1), and the templating its steps use. + +A journey file is YAML, one journey per file, committed to the app's repo: + + version: 1 + name: checkout + type: checkout + steps: + - name: log in + request: {method: POST, path: /api/login, + json: {email: "{{unique.email}}", password: "{{env.SHOP_PASSWORD}}"}} + expect: {status: 200} + extract: {token: body.token} + - name: place order + request: {method: POST, path: /api/orders, + headers: {Authorization: "Bearer {{token}}"}} + expect: {status: [200, 201]} + +Placeholders: `{{name}}` is a value an earlier step extracted; `{{unique.email}}` +and `{{unique.id}}` are fresh for each run of the journey; `{{env.NAME}}` reads an +environment variable, which is how secrets stay out of the file. Everything here +is pure: loading never touches the network. +""" + +from __future__ import annotations + +import re +import secrets +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +import yaml + +FORMAT_VERSION = 1 +DEFAULT_DIR = Path("prescale") / "journeys" + +# Journey types PreScale replays, and each one's default share of the traffic mix. +TYPES: dict[str, int] = { + "browse": 60, "search": 25, "account": 8, "checkout": 5, "write": 5, "auth": 2, +} +# Types PreScale never replays: they message real people or destroy data. +REFUSED_TYPES = { + "side_effect": "it sends email, SMS or invites to real people", + "destructive": "it deletes data or changes admin settings", +} +METHODS = {"GET", "HEAD", "OPTIONS", "POST", "PUT", "PATCH", "DELETE"} +READ_METHODS = {"GET", "HEAD", "OPTIONS"} + +_PLACEHOLDER = re.compile(r"\{\{\s*([A-Za-z_][\w.]*)\s*\}\}") +_SENSITIVE_KEY = re.compile(r"pass(word|wd)?|secret|token|api[-_]?key|authorization|cookie", + re.IGNORECASE) +_UNIQUE = {"email", "id"} +_SOURCES = ("body", "header") + + +class JourneyError(Exception): + """A journey file that can't be used, with a message that says how to fix it.""" + + +@dataclass +class Step: + name: str + method: str + path: str + headers: dict[str, str] = field(default_factory=dict) + json: Any = None + body: str | None = None + expect: list[int] | None = None # None: any status below 400 + extract: dict[str, str] = field(default_factory=dict) + think: float = 0.0 + + @property + def label(self) -> str: + """The route as reports show it: method plus path, placeholders kept.""" + return f"{self.method} {self.path.split('?')[0]}" + + def accepts(self, status: int) -> bool: + if self.expect is None: + return status < 400 + return status in self.expect + + +@dataclass +class Journey: + name: str + type: str + weight: float + steps: list[Step] + source: Path | None = None + + @property + def writes(self) -> bool: + """Whether replaying it can change data on the target.""" + return any(step.method not in READ_METHODS for step in self.steps) + + def env_names(self) -> set[str]: + """Environment variables its steps read.""" + return {name[4:] for step in self.steps for name in _names(_step_values(step)) + if name.startswith("env.")} + + +# --- loading -------------------------------------------------------------- + +def load_dir(directory: Path = DEFAULT_DIR) -> tuple[list[Journey], list[str]]: + """Every journey in DIRECTORY, sorted by name, plus one error message per file + that couldn't be loaded. A missing directory is no journeys, not an error.""" + if not directory.is_dir(): + return [], [] + journeys: list[Journey] = [] + errors: list[str] = [] + for path in sorted([*directory.glob("*.yaml"), *directory.glob("*.yml")]): + try: + journeys.append(load_file(path)) + except JourneyError as exc: + errors.append(f"{path}: {exc}") + seen: dict[str, Path] = {} + for journey in journeys: + if journey.name in seen: + errors.append(f"{journey.source}: the name '{journey.name}' is already used by " + f"{seen[journey.name]}") + seen.setdefault(journey.name, journey.source) + return sorted(journeys, key=lambda j: j.name), errors + + +def load_file(path: Path) -> Journey: + try: + data = yaml.safe_load(path.read_text()) + except yaml.YAMLError as exc: + raise JourneyError(f"not valid YAML ({exc})") from exc + journey = parse(data) + journey.source = path + return journey + + +def parse(data: Any) -> Journey: + """Check a decoded journey document and build a `Journey`.""" + if not isinstance(data, dict): + raise JourneyError("a journey file must be a mapping with version, name and steps") + _only(data, {"version", "name", "type", "weight", "steps"}, "the journey") + if data.get("version") != FORMAT_VERSION: + raise JourneyError(f"version must be {FORMAT_VERSION} (got {data.get('version')!r})") + name = data.get("name") + if not isinstance(name, str) or not re.fullmatch(r"[a-z0-9][a-z0-9_-]*", name): + raise JourneyError("name must be lowercase letters, digits, - or _ (e.g. checkout)") + jtype = data.get("type", "browse") + if jtype in REFUSED_TYPES: + raise JourneyError(f"type '{jtype}' is never replayed: {REFUSED_TYPES[jtype]}") + if jtype not in TYPES: + raise JourneyError(f"type must be one of {', '.join(TYPES)} (got {jtype!r})") + weight = data.get("weight", TYPES[jtype]) + if isinstance(weight, bool) or not isinstance(weight, (int, float)) or weight <= 0: + raise JourneyError("weight must be a positive number") + raw_steps = data.get("steps") + if not isinstance(raw_steps, list) or not raw_steps: + raise JourneyError("steps must be a non-empty list") + + steps = [_parse_step(raw, i) for i, raw in enumerate(raw_steps, 1)] + _check_placeholders(steps) + return Journey(name=name, type=jtype, weight=float(weight), steps=steps) + + +def _parse_step(raw: Any, index: int) -> Step: + where = f"step {index}" + if not isinstance(raw, dict): + raise JourneyError(f"{where} must be a mapping with name and request") + _only(raw, {"name", "request", "expect", "extract", "think"}, where) + name = raw.get("name") or f"step {index}" + if not isinstance(name, str): + raise JourneyError(f"{where}: name must be text") + where = f"step {index} ({name})" + + request = raw.get("request") + if not isinstance(request, dict): + raise JourneyError(f"{where}: request must be a mapping with method and path") + _only(request, {"method", "path", "headers", "json", "body"}, f"{where} request") + method = str(request.get("method", "GET")).upper() + if method not in METHODS: + raise JourneyError(f"{where}: method {method} isn't supported") + path = request.get("path") + if not isinstance(path, str) or not path.startswith("/"): + raise JourneyError(f"{where}: path must start with / (journeys only call the app " + "they're run against, never another site)") + headers = request.get("headers") or {} + if not isinstance(headers, dict) or not all(isinstance(v, str) for v in headers.values()): + raise JourneyError(f"{where}: headers must map names to text") + if "json" in request and "body" in request: + raise JourneyError(f"{where}: give either json or body, not both") + body = request.get("body") + if body is not None and not isinstance(body, str): + raise JourneyError(f"{where}: body must be text (use json for structured data)") + _check_secrets({"headers": headers, "json": request.get("json")}, where) + + return Step( + name=name, method=method, path=path, + headers={str(k): v for k, v in headers.items()}, + json=request.get("json"), body=body, + expect=_parse_expect(raw.get("expect"), where), + extract=_parse_extract(raw.get("extract"), where), + think=_parse_seconds(raw.get("think", 0), where), + ) + + +def _parse_expect(raw: Any, where: str) -> list[int] | None: + if raw is None: + return None + if not isinstance(raw, dict) or set(raw) != {"status"}: + raise JourneyError(f"{where}: expect must be {{status: 200}} or {{status: [200, 201]}}") + status = raw["status"] + codes = status if isinstance(status, list) else [status] + if not codes or not all(isinstance(c, int) and 100 <= c <= 599 for c in codes): + raise JourneyError(f"{where}: expect status must be HTTP status codes") + return codes + + +def _parse_extract(raw: Any, where: str) -> dict[str, str]: + if raw is None: + return {} + if not isinstance(raw, dict): + raise JourneyError(f"{where}: extract must map names to body. or header.") + for name, source in raw.items(): + if not isinstance(name, str) or not re.fullmatch(r"[A-Za-z_]\w*", name) \ + or name in {"env", "unique"}: + raise JourneyError(f"{where}: can't extract into {name!r}") + if not isinstance(source, str) or source.split(".", 1)[0] not in _SOURCES \ + or "." not in source: + raise JourneyError(f"{where}: extract {name} must come from body. " + "or header.") + return dict(raw) + + +def _parse_seconds(raw: Any, where: str) -> float: + if isinstance(raw, bool): + raise JourneyError(f"{where}: think must be seconds, e.g. 2 or 2s or 500ms") + if isinstance(raw, (int, float)) and raw >= 0: + return float(raw) + match = re.fullmatch(r"(\d+(?:\.\d+)?)(ms|s)", str(raw).strip()) + if not match: + raise JourneyError(f"{where}: think must be seconds, e.g. 2 or 2s or 500ms") + value = float(match.group(1)) + return value / 1000 if match.group(2) == "ms" else value + + +def _only(data: dict, allowed: set[str], where: str) -> None: + unknown = sorted(set(data) - allowed) + if unknown: + raise JourneyError(f"{where} has unknown field(s): {', '.join(map(str, unknown))}") + + +def _check_secrets(value: Any, where: str, key: str = "") -> None: + """Refuse a literal value under a secret-looking key: secrets come from {{env.*}} + (or an earlier step's extract), so they never end up committed in the file.""" + if isinstance(value, dict): + for k, v in value.items(): + _check_secrets(v, where, str(k)) + elif isinstance(value, list): + for v in value: + _check_secrets(v, where, key) + elif isinstance(value, str) and key and _SENSITIVE_KEY.search(key) \ + and not _PLACEHOLDER.search(value): + raise JourneyError(f"{where}: {key} has a literal value. Put secrets in an " + f"environment variable and write \"{{{{env.NAME}}}}\" instead") + + +def _check_placeholders(steps: list[Step]) -> None: + """Every {{name}} must be extracted by an earlier step, or be env./unique.""" + known: set[str] = set() + for i, step in enumerate(steps, 1): + for name in _names(_step_values(step)): + if name.startswith("env."): + continue + if name.startswith("unique."): + if name[7:] not in _UNIQUE: + raise JourneyError(f"step {i} ({step.name}): {{{{{name}}}}} isn't known; " + "use unique.email or unique.id") + continue + if name not in known: + raise JourneyError(f"step {i} ({step.name}) uses {{{{{name}}}}}, but no " + "earlier step extracts it") + known.update(step.extract) + + +def _step_values(step: Step) -> list[Any]: + return [step.path, step.headers, step.json, step.body] + + +def _names(value: Any) -> list[str]: + if isinstance(value, str): + return _PLACEHOLDER.findall(value) + if isinstance(value, dict): + return [n for k, v in value.items() for n in (*_names(k), *_names(v))] + if isinstance(value, list): + return [n for v in value for n in _names(v)] + return [] + + +# --- templating ----------------------------------------------------------- + +class TemplateError(Exception): + """A placeholder with no value at replay time (e.g. an unset env variable).""" + + +def fresh_uniques() -> dict[str, str]: + """Values for {{unique.*}}, new for each run of a journey.""" + token = secrets.token_hex(6) + # example.com is reserved for examples (RFC 2606): mail to it goes nowhere. + return {"unique.email": f"prescale+{token}@example.com", "unique.id": token} + + +def render(value: Any, variables: dict[str, Any], env: dict[str, str]) -> Any: + """Fill placeholders in a step value. A string that is exactly one placeholder + keeps the value's type (an extracted id stays a number in JSON).""" + if isinstance(value, str): + whole = _PLACEHOLDER.fullmatch(value.strip()) + if whole: + return _lookup(whole.group(1), variables, env) + return _PLACEHOLDER.sub(lambda m: str(_lookup(m.group(1), variables, env)), value) + if isinstance(value, dict): + return {render(k, variables, env): render(v, variables, env) for k, v in value.items()} + if isinstance(value, list): + return [render(v, variables, env) for v in value] + return value + + +def _lookup(name: str, variables: dict[str, Any], env: dict[str, str]) -> Any: + if name.startswith("env."): + if name[4:] not in env: + raise TemplateError(f"environment variable {name[4:]} isn't set") + return env[name[4:]] + if name not in variables: + raise TemplateError(f"{{{{{name}}}}} has no value") + return variables[name] diff --git a/cli/src/prescale_cli/journeys/replay.py b/cli/src/prescale_cli/journeys/replay.py new file mode 100644 index 0000000..b6e67ba --- /dev/null +++ b/cli/src/prescale_cli/journeys/replay.py @@ -0,0 +1,133 @@ +"""Run a journey's steps against an app. + +`check` replays a journey once, as a single user, to see whether it still works +against the app as it is now. A step that fails its expectation here, with no +load at all, means the journey is stale (the app changed), not that the app +broke under load. Each run gets its own client, so cookies and extracted values +never leak between users. +""" + +from __future__ import annotations + +import os +import time +from dataclasses import dataclass, field +from typing import Any +from urllib.parse import urljoin + +import httpx + +from prescale_cli.journeys.model import Journey, Step, TemplateError, fresh_uniques, render +from prescale_cli.loadtest import client_headers + +_SNIPPET = 160 + + +@dataclass +class StepCheck: + step: Step + status: int | None = None + latency: float = 0.0 + ok: bool = False + problem: str | None = None # why it failed, in plain words + + +@dataclass +class JourneyCheck: + journey: Journey + steps: list[StepCheck] = field(default_factory=list) + + @property + def ok(self) -> bool: + return len(self.steps) == len(self.journey.steps) and all(s.ok for s in self.steps) + + @property + def failed(self) -> StepCheck | None: + return next((s for s in self.steps if not s.ok), None) + + @property + def stale(self) -> bool: + """The app answered, but not the way the journey expects.""" + failed = self.failed + return failed is not None and failed.status is not None + + +async def check(journey: Journey, base_url: str, *, timeout: float = 10.0, + headers: dict[str, str] | None = None, env: dict[str, str] | None = None, + transport: httpx.AsyncBaseTransport | None = None) -> JourneyCheck: + """Replay JOURNEY once against BASE_URL and stop at the first step that fails.""" + env = dict(os.environ) if env is None else env + result = JourneyCheck(journey) + variables: dict[str, Any] = fresh_uniques() + async with httpx.AsyncClient(timeout=timeout, follow_redirects=True, + headers=client_headers(headers), transport=transport) as client: + for step in journey.steps: + outcome = await run_step(client, step, base_url, variables, env) + result.steps.append(outcome) + if not outcome.ok: + break + return result + + +async def run_step(client: httpx.AsyncClient, step: Step, base_url: str, + variables: dict[str, Any], env: dict[str, str]) -> StepCheck: + """Send one step, judge it against `expect`, and pull its `extract` values + into VARIABLES for the steps after it.""" + outcome = StepCheck(step) + try: + path = render(step.path, variables, env) + req_headers = render(step.headers, variables, env) + payload = render(step.json, variables, env) if step.json is not None else None + body = render(step.body, variables, env) if step.body is not None else None + except TemplateError as exc: + outcome.problem = str(exc) + return outcome + + start = time.perf_counter() + try: + resp = await client.request(step.method, urljoin(base_url, path), + headers=req_headers, json=payload, content=body) + except httpx.TimeoutException: + outcome.problem = "timed out" + return outcome + except httpx.HTTPError as exc: + outcome.problem = f"couldn't reach the app ({exc.__class__.__name__})" + return outcome + outcome.latency = time.perf_counter() - start + outcome.status = resp.status_code + + if not step.accepts(resp.status_code): + wanted = (" or ".join(map(str, step.expect)) if step.expect else "a status below 400") + outcome.problem = f"returned {resp.status_code}, expected {wanted}" + snippet = resp.text.strip().replace("\n", " ")[:_SNIPPET] + if snippet: + outcome.problem += f": {snippet}" + return outcome + + for name, source in step.extract.items(): + value = extract(resp, source) + if value is None: + outcome.problem = f"couldn't find {source} in the response to extract {name}" + return outcome + variables[name] = value + outcome.ok = True + return outcome + + +def extract(resp: httpx.Response, source: str) -> Any: + """Read `body.a.b.0` (JSON) or `header.Name` from a response; None if missing.""" + kind, _, where = source.partition(".") + if kind == "header": + return resp.headers.get(where) + try: + value: Any = resp.json() + except ValueError: + return None + for part in where.split("."): + if isinstance(value, dict) and part in value: + value = value[part] + elif isinstance(value, list) and part.isdigit() and int(part) < len(value): + value = value[int(part)] + else: + return None + return value diff --git a/cli/src/prescale_cli/main.py b/cli/src/prescale_cli/main.py index d87c4c5..67f66a8 100644 --- a/cli/src/prescale_cli/main.py +++ b/cli/src/prescale_cli/main.py @@ -9,6 +9,7 @@ compare, history, investigate, + journeys, login, mcp, profiles, @@ -33,6 +34,7 @@ def cli() -> None: prescale run http://localhost:8000 # load test: what breaks first prescale investigate http://localhost:8000 # ...and why, plus how to fix it prescale audit https://myapp.com # static scaling-hygiene check + prescale journeys verify http://localhost:8000 # check your journeys still replay prescale history # list saved runs prescale compare # capacity diff of the last two runs prescale profiles # launch scenarios for --profile @@ -45,6 +47,7 @@ def cli() -> None: cli.add_command(run.run) cli.add_command(investigate.investigate) +cli.add_command(journeys.journeys) cli.add_command(audit.audit) cli.add_command(history.history) cli.add_command(compare.compare) diff --git a/cli/tests/test_journeys.py b/cli/tests/test_journeys.py new file mode 100644 index 0000000..68c8142 --- /dev/null +++ b/cli/tests/test_journeys.py @@ -0,0 +1,260 @@ +"""Journey files: the format, its safety rules, and replaying a journey once.""" + +import asyncio +import json + +import httpx +import pytest +from click.testing import CliRunner + +from prescale_cli.journeys.model import ( + JourneyError, + TemplateError, + load_dir, + parse, + render, +) +from prescale_cli.journeys.replay import check +from prescale_cli.main import cli + +CHECKOUT = { + "version": 1, + "name": "checkout", + "type": "checkout", + "steps": [ + {"name": "log in", + "request": {"method": "POST", "path": "/api/login", + "json": {"email": "{{unique.email}}", "password": "{{env.PW}}"}}, + "expect": {"status": 200}, + "extract": {"token": "body.token"}}, + {"name": "add to cart", + "request": {"method": "POST", "path": "/api/cart", + "headers": {"Authorization": "Bearer {{token}}"}}, + "expect": {"status": [200, 201]}, + "extract": {"cart_id": "body.id"}}, + {"name": "place order", + "request": {"method": "POST", "path": "/api/carts/{{cart_id}}/order", + "headers": {"Authorization": "Bearer {{token}}"}}, + "think": "500ms"}, + ], +} + + +def _with(**changes): + doc = json.loads(json.dumps(CHECKOUT)) + doc.update(changes) + return doc + + +def _step(i, **changes): + doc = json.loads(json.dumps(CHECKOUT)) + doc["steps"][i].update(changes) + return doc + + +# --- the format ------------------------------------------------------------- + +def test_parse_valid_journey(): + j = parse(CHECKOUT) + assert j.name == "checkout" and j.type == "checkout" + assert j.weight == 5 # the type's default share + assert [s.label for s in j.steps] == [ + "POST /api/login", "POST /api/cart", "POST /api/carts/{{cart_id}}/order"] + assert j.steps[1].expect == [200, 201] + assert j.steps[2].expect is None and j.steps[2].think == 0.5 + assert j.writes + assert j.env_names() == {"PW"} + + +def test_read_only_journey_does_not_write(): + j = parse({"version": 1, "name": "browse", "steps": [{"request": {"path": "/"}}]}) + assert j.type == "browse" and not j.writes + assert j.steps[0].method == "GET" and j.steps[0].name == "step 1" + + +@pytest.mark.parametrize("doc, message", [ + (_with(version=2), "version must be 1"), + (_with(name="Check Out"), "name must be lowercase"), + (_with(type="spam"), "type must be one of"), + (_with(weight=0), "weight must be a positive number"), + (_with(steps=[]), "steps must be a non-empty list"), + (_with(extra=True), "unknown field(s): extra"), + (_step(0, request={"method": "POST", "path": "https://evil.example/x"}), "path must start with /"), + (_step(0, request={"method": "BREW", "path": "/"}), "method BREW isn't supported"), + (_step(0, expect={"status": "ok"}), "HTTP status codes"), + (_step(0, extract={"token": "cookie.sid"}), "body. or header."), + (_step(0, think="soon"), "think must be seconds"), +]) +def test_parse_rejects_bad_journeys(doc, message): + with pytest.raises(JourneyError, match=message.replace("(", r"\(").replace(")", r"\)")): + parse(doc) + + +@pytest.mark.parametrize("jtype", ["side_effect", "destructive"]) +def test_refused_types_are_never_loaded(jtype): + with pytest.raises(JourneyError, match="never replayed"): + parse(_with(type=jtype)) + + +def test_placeholder_must_be_extracted_by_an_earlier_step(): + doc = json.loads(json.dumps(CHECKOUT)) + del doc["steps"][0]["extract"] + with pytest.raises(JourneyError, match=r"uses \{\{token\}\}, but no earlier step"): + parse(doc) + + +def test_unknown_unique_placeholder_is_rejected(): + doc = _step(0, request={"method": "GET", "path": "/u/{{unique.phone}}"}) + with pytest.raises(JourneyError, match="unique.email or unique.id"): + parse(doc) + + +@pytest.mark.parametrize("request_part", [ + {"json": {"email": "a@b.co", "password": "hunter2"}}, + {"headers": {"Authorization": "Bearer abc123"}}, + {"json": {"user": {"api_key": "sk-live"}}}, +]) +def test_literal_secrets_are_refused(request_part): + doc = _step(0, request={"method": "POST", "path": "/api/login", **request_part}) + with pytest.raises(JourneyError, match="literal value"): + parse(doc) + + +def test_load_dir_reports_bad_files_and_duplicate_names(tmp_path): + (tmp_path / "a.yaml").write_text(json.dumps(CHECKOUT)) + (tmp_path / "b.yml").write_text(json.dumps(CHECKOUT)) # same name + (tmp_path / "c.yaml").write_text("version: 1\nname: [oops\n") # not YAML + found, errors = load_dir(tmp_path) + assert [j.name for j in found] == ["checkout", "checkout"] + assert any("not valid YAML" in e for e in errors) + assert any("already used" in e for e in errors) + + +def test_load_dir_missing_directory_is_empty(tmp_path): + assert load_dir(tmp_path / "nope") == ([], []) + + +# --- templating ------------------------------------------------------------- + +def test_render_keeps_type_of_whole_placeholder(): + variables = {"id": 7, "token": "t1"} + assert render({"cart": "{{id}}"}, variables, {}) == {"cart": 7} + assert render("/carts/{{id}}/order", variables, {}) == "/carts/7/order" + assert render("Bearer {{token}}", variables, {}) == "Bearer t1" + assert render("{{env.PW}}", {}, {"PW": "s3cret"}) == "s3cret" + + +def test_render_missing_env_is_an_error(): + with pytest.raises(TemplateError, match="PW isn't set"): + render("{{env.PW}}", {}, {}) + + +# --- replaying once ---------------------------------------------------------- + +def _shop(order_status=201, cart_path="/api/cart"): + seen = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(request) + if request.url.path == "/api/login": + body = json.loads(request.content) + assert body["password"] == "pw" and body["email"].endswith("@example.com") + return httpx.Response(200, json={"token": "t1"}) + assert request.headers["authorization"] == "Bearer t1" + if request.url.path == cart_path: + return httpx.Response(201, json={"id": 42}) + if request.url.path == "/api/carts/42/order": + return httpx.Response(order_status, text="pool exhausted") + return httpx.Response(404, json={"error": "not found"}) + + return httpx.MockTransport(handler), seen + + +def _check(transport): + return asyncio.run(check(parse(CHECKOUT), "http://shop.test/", env={"PW": "pw"}, + transport=transport)) + + +def test_check_replays_and_carries_extracted_values(): + transport, seen = _shop() + result = _check(transport) + assert result.ok and not result.stale + assert [s.status for s in result.steps] == [200, 201, 201] + assert str(seen[2].url) == "http://shop.test/api/carts/42/order" + + +def test_check_marks_a_changed_app_as_stale(): + transport, _ = _shop(cart_path="/api/basket") + result = _check(transport) + assert not result.ok and result.stale + assert len(result.steps) == 2 # stops at the failing step + assert result.failed.problem.startswith("returned 404, expected 200 or 201") + + +def test_check_without_expect_fails_on_4xx_and_5xx(): + transport, _ = _shop(order_status=500) + result = _check(transport) + assert result.stale + assert "expected a status below 400: pool exhausted" in result.failed.problem + + +def test_check_unreachable_app_is_not_stale(): + def boom(request): + raise httpx.ConnectError("refused") + result = _check(httpx.MockTransport(boom)) + assert not result.ok and not result.stale + assert result.failed.problem.startswith("couldn't reach the app") + + +def test_check_missing_extract_fails_the_step(): + def handler(request): + return httpx.Response(200, json={"no_token": True}) + result = _check(httpx.MockTransport(handler)) + assert "couldn't find body.token" in result.failed.problem + + +# --- the command ------------------------------------------------------------- + +def _out(res): + return " ".join(res.output.split()) # rich wraps long lines + + +def _write(tmp_path, doc=CHECKOUT, name="checkout.yaml"): + (tmp_path / name).write_text(json.dumps(doc)) + return tmp_path + + +def test_journeys_lists_files(tmp_path): + res = CliRunner().invoke(cli, ["journeys", "--dir", str(_write(tmp_path))]) + assert res.exit_code == 0, res.output + assert "checkout" in res.output and "yes" in res.output + + +def test_verify_refuses_writes_on_a_non_local_host(tmp_path): + res = CliRunner().invoke(cli, ["journeys", "verify", "https://shop.example.com", + "--dir", str(_write(tmp_path))], env={"PW": "pw"}) + assert res.exit_code == 2 + assert "--allow-writes shop.example.com" in _out(res) + + +def test_verify_refuses_when_env_is_missing(tmp_path): + res = CliRunner().invoke(cli, ["journeys", "verify", "http://localhost:1", + "--dir", str(_write(tmp_path))], env={"PW": None}) + assert res.exit_code == 2 + assert "aren't set: PW" in _out(res) + + +def test_verify_reports_invalid_files(tmp_path): + bad = _with(name="Bad Name") + res = CliRunner().invoke(cli, ["journeys", "verify", "http://localhost:1", + "--dir", str(_write(tmp_path, bad))]) + assert res.exit_code == 2 + assert "name must be lowercase" in _out(res) + + +def test_verify_unknown_journey_name(tmp_path): + res = CliRunner().invoke(cli, ["journeys", "verify", "http://localhost:1", + "--dir", str(_write(tmp_path)), "--journey", "nope"], + env={"PW": "pw"}) + assert res.exit_code == 2 + assert "no journey named nope" in _out(res) diff --git a/examples/README.md b/examples/README.md index 175dd94..bdd75ef 100644 --- a/examples/README.md +++ b/examples/README.md @@ -17,3 +17,12 @@ PreScale reports that `/checkout` breaks first, then diagnoses it as a connection-pool ceiling and suggests raising the pool / adding pgbouncer. It's the app behind the demo in the README. Tune the `Semaphore(25)` to move the breaking point. + +### Journeys + +The shop also has a small JSON API (log in, add to cart, place an order; the order +draws from the same pool) and two journeys for it in `journeys/`: + +```bash +SHOP_PASSWORD=demo prescale journeys verify http://localhost:8400 --dir examples/journeys +``` diff --git a/examples/fragile-shop.py b/examples/fragile-shop.py index 0edb2af..2a6bd2d 100644 --- a/examples/fragile-shop.py +++ b/examples/fragile-shop.py @@ -7,10 +7,18 @@ python examples/fragile-shop.py 8400 prescale investigate http://localhost:8400 --path /search --path /product --path /checkout + +It also has a small JSON API for journeys (log in, add to cart, place an order; +the order draws from the same pool), with journey files in examples/journeys/: + + SHOP_PASSWORD=demo prescale journeys verify http://localhost:8400 --dir examples/journeys """ import asyncio +import itertools +import json import random +import secrets import sys PAGE = (b"" @@ -18,7 +26,13 @@ b'' b"acme.shop") -_REASON = {200: "OK", 404: "Not Found", 500: "Internal Server Error"} +_REASON = {200: "OK", 201: "Created", 400: "Bad Request", 401: "Unauthorized", + 404: "Not Found", 500: "Internal Server Error"} + +PRODUCTS = [{"id": 1, "name": "mug"}, {"id": 2, "name": "tee"}, {"id": 3, "name": "cap"}] +_tokens: set[str] = set() +_carts: dict[int, str] = {} # cart id -> owner token +_cart_ids = itertools.count(1) def _resp(code, body=b"ok", ctype="text/plain"): @@ -36,13 +50,21 @@ async def _handle(reader, writer, pool): if not line: return parts = line.decode(errors="replace").split() + method = parts[0] if parts else "GET" path = parts[1].split("?")[0] if len(parts) >= 2 else "/" - while True: # drain request headers + headers = {} + while True: # read request headers header = await reader.readline() if header in (b"\r\n", b"\n", b""): break + name, _, value = header.decode(errors="replace").partition(":") + headers[name.strip().lower()] = value.strip() + length = int(headers.get("content-length") or 0) + body = await reader.readexactly(length) if length else b"" - if path == "/": + if path.startswith("/api/"): + out = await _api(method, path, headers, body, pool) + elif path == "/": await asyncio.sleep(0.005) out = _resp(200, PAGE, "text/html") elif path in ("/app.css", "/app.js"): @@ -75,6 +97,49 @@ async def _handle(reader, writer, pool): pass +def _json(code, data): + return _resp(code, json.dumps(data).encode(), "application/json") + + +async def _api(method, path, headers, body, pool): + """The shop's JSON API, used by the example journeys.""" + token = headers.get("authorization", "").removeprefix("Bearer ").strip() + if method == "GET" and path == "/api/products": + await asyncio.sleep(random.uniform(0.005, 0.02)) + return _json(200, PRODUCTS) + if method == "POST" and path == "/api/login": + try: + creds = json.loads(body or b"{}") + except ValueError: + return _json(400, {"error": "send JSON"}) + if not creds.get("email") or not creds.get("password"): + return _json(401, {"error": "email and password required"}) + await asyncio.sleep(0.02) # password hashing + new = secrets.token_hex(8) + _tokens.add(new) + return _json(200, {"token": new}) + if token not in _tokens: + return _json(401, {"error": "log in first"}) + if method == "POST" and path == "/api/cart": + cart = next(_cart_ids) + _carts[cart] = token + return _json(201, {"id": cart}) + if method == "POST" and path.startswith("/api/carts/") and path.endswith("/order"): + cart = path.split("/")[3] + if not cart.isdigit() or _carts.get(int(cart)) != token: + return _json(404, {"error": "no such cart"}) + try: # grab a DB connection + await asyncio.wait_for(pool.acquire(), timeout=0.001) + except asyncio.TimeoutError: + return _json(500, {"error": "could not acquire DB connection (pool exhausted)"}) + try: + await asyncio.sleep(random.uniform(0.08, 0.12)) + return _json(201, {"order": f"o-{cart}"}) + finally: + pool.release() + return _json(404, {"error": "not found"}) + + async def main(port): pool = asyncio.Semaphore(25) # 25 DB connections server = await asyncio.start_server( diff --git a/examples/journeys/browse.yaml b/examples/journeys/browse.yaml new file mode 100644 index 0000000..b10e145 --- /dev/null +++ b/examples/journeys/browse.yaml @@ -0,0 +1,12 @@ +version: 1 +name: browse +type: browse +steps: + - name: home + request: {method: GET, path: /} + - name: list products + request: {method: GET, path: /api/products} + expect: {status: 200} + extract: {product_id: body.0.id} + - name: product page + request: {method: GET, path: /product} diff --git a/examples/journeys/checkout.yaml b/examples/journeys/checkout.yaml new file mode 100644 index 0000000..6725e2b --- /dev/null +++ b/examples/journeys/checkout.yaml @@ -0,0 +1,25 @@ +version: 1 +name: checkout +type: checkout +steps: + - name: log in + request: + method: POST + path: /api/login + json: {email: "{{unique.email}}", password: "{{env.SHOP_PASSWORD}}"} + expect: {status: 200} + extract: {token: body.token} + - name: add to cart + request: + method: POST + path: /api/cart + headers: {Authorization: "Bearer {{token}}"} + expect: {status: 201} + extract: {cart_id: body.id} + - name: place order + request: + method: POST + path: "/api/carts/{{cart_id}}/order" + headers: {Authorization: "Bearer {{token}}"} + expect: {status: 201} + think: 1s From c3b3c62be8b1d30624225a4cee772f4d87cb50f3 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 2 Nov 2026 14:07:32 +0000 Subject: [PATCH 2/2] fix(journeys): keep every request on the app, tighten the secrets rule --- README.md | 2 +- cli/src/prescale_cli/commands/journeys.py | 19 ++-- cli/src/prescale_cli/journeys/model.py | 43 +++++++--- cli/src/prescale_cli/journeys/replay.py | 48 +++++++++-- cli/tests/test_journeys.py | 100 +++++++++++++++++++++- examples/fragile-shop.py | 12 ++- 6 files changed, 193 insertions(+), 31 deletions(-) diff --git a/README.md b/README.md index 552c8a1..7be10e9 100644 --- a/README.md +++ b/README.md @@ -200,7 +200,7 @@ prescale journeys verify http://localhost:3000 # replay each once, as one us - `{{name}}` is a value an earlier step pulled out with `extract` (`body.` or `header.`); `{{unique.email}}` and `{{unique.id}}` are fresh each run; `{{env.NAME}}` reads an environment variable. - **Secrets never go in the file.** A literal password, token, API key or `Authorization` value is refused; use `{{env.NAME}}`. -- Paths must start with `/`: a journey only calls the app it's run against. +- Paths start with `/` and are relative to the URL you give (`http://host/app` + `/api/cart` is `http://host/app/api/cart`). A journey only calls that app: a request or redirect to another host fails the step. - Journeys that change data (POST, PUT, PATCH, DELETE) run against local hosts, or a host you name with `--allow-writes staging.myapp.com`. Types `side_effect` (emails, SMS, invites) and `destructive` are never replayed. Try it on the example shop: `python examples/fragile-shop.py 8400`, then `SHOP_PASSWORD=demo prescale journeys verify http://localhost:8400 --dir examples/journeys`. diff --git a/cli/src/prescale_cli/commands/journeys.py b/cli/src/prescale_cli/commands/journeys.py index 626b39a..23f112a 100644 --- a/cli/src/prescale_cli/commands/journeys.py +++ b/cli/src/prescale_cli/commands/journeys.py @@ -48,7 +48,7 @@ def journeys(ctx: click.Context, directory: Path) -> None: table.add_column(col) for j in found: table.add_row(j.name, j.type, f"{j.weight:g}", str(len(j.steps)), - "yes" if j.writes else "no", str(j.source)) + "yes" if j.writes else "no", escape(str(j.source))) console.print(table) @@ -94,8 +94,7 @@ def verify(url: str, directory: Path, only: tuple[str, ...], allow_writes: tuple f"aren't set: {', '.join(unset)}.") raise SystemExit(2) - base = url if url.endswith("/") else url + "/" - results = asyncio.run(_check_all(found, base, timeout=timeout, headers=headers)) + results = asyncio.run(_check_all(found, url, timeout=timeout, headers=headers)) for result in results: _print_check(result) @@ -120,7 +119,8 @@ def _load_or_exit(directory: Path) -> list[Journey]: if errors: raise SystemExit(2) if not found: - console.print(f"No journeys in {directory}. Add one as {directory}/.yaml.") + console.print(f"No journeys in {escape(str(directory))}. Add one as " + f"{escape(str(directory))}/.yaml.", highlight=False) raise SystemExit(1) return found @@ -147,17 +147,18 @@ def _print_check(result: JourneyCheck) -> None: journey = result.journey mark = "[green]✓[/green]" if result.ok else ("[yellow]stale[/yellow]" if result.stale else "[red]✗[/red]") - console.print(f"\n{mark} [bold]{journey.name}[/bold] [dim]({journey.type}, " + console.print(f"\n{mark} [bold]{escape(journey.name)}[/bold] [dim]({journey.type}, " f"{len(journey.steps)} steps)[/dim]") for i, step in enumerate(journey.steps, 1): done = result.steps[i - 1] if i <= len(result.steps) else None + name, label = escape(step.name), escape(step.label) if done is None: - console.print(f" [dim]{i}. {step.name} {step.label} (not run)[/dim]") + console.print(f" [dim]{i}. {name} {label} (not run)[/dim]", highlight=False) elif done.ok: - console.print(f" {i}. {step.name} [dim]{step.label}[/dim] " - f"{done.status} {done.latency * 1000:.0f}ms") + console.print(f" {i}. {name} [dim]{label}[/dim] " + f"{done.status} {done.latency * 1000:.0f}ms", highlight=False) else: - console.print(f" {i}. {step.name} [dim]{step.label}[/dim] " + console.print(f" {i}. {name} [dim]{label}[/dim] " f"[red]{escape(done.problem or '')}[/red]", highlight=False) if done.status is not None and done.status >= 500: console.print(" [dim]A 5xx with no load is usually a bug in the app, " diff --git a/cli/src/prescale_cli/journeys/model.py b/cli/src/prescale_cli/journeys/model.py index 1ca708b..699133f 100644 --- a/cli/src/prescale_cli/journeys/model.py +++ b/cli/src/prescale_cli/journeys/model.py @@ -24,11 +24,14 @@ from __future__ import annotations +import json +import math import re import secrets from dataclasses import dataclass, field from pathlib import Path from typing import Any +from urllib.parse import parse_qsl, urlsplit import yaml @@ -48,8 +51,14 @@ READ_METHODS = {"GET", "HEAD", "OPTIONS"} _PLACEHOLDER = re.compile(r"\{\{\s*([A-Za-z_][\w.]*)\s*\}\}") -_SENSITIVE_KEY = re.compile(r"pass(word|wd)?|secret|token|api[-_]?key|authorization|cookie", - re.IGNORECASE) +# A key is secret-looking when it is, or ends with, one of these words: +# password, client_secret, x-api-key, access_token, Authorization, Cookie. +_SENSITIVE_KEY = re.compile( + r"(^|[-_.])(password|passwd|pwd|secret|token|api[-_]?key|authorization|cookie)$", + re.IGNORECASE) +# The secret itself must be a placeholder, optionally after an auth scheme. +_SECRET_OK = re.compile(r"((Bearer|Basic|Token)\s+)?\{\{\s*[A-Za-z_][\w.]*\s*\}\}", + re.IGNORECASE) _UNIQUE = {"email", "id"} _SOURCES = ("body", "header") @@ -149,7 +158,8 @@ def parse(data: Any) -> Journey: if jtype not in TYPES: raise JourneyError(f"type must be one of {', '.join(TYPES)} (got {jtype!r})") weight = data.get("weight", TYPES[jtype]) - if isinstance(weight, bool) or not isinstance(weight, (int, float)) or weight <= 0: + if isinstance(weight, bool) or not isinstance(weight, (int, float)) \ + or not math.isfinite(weight) or weight <= 0: raise JourneyError("weight must be a positive number") raw_steps = data.get("steps") if not isinstance(raw_steps, list) or not raw_steps: @@ -168,7 +178,8 @@ def _parse_step(raw: Any, index: int) -> Step: name = raw.get("name") or f"step {index}" if not isinstance(name, str): raise JourneyError(f"{where}: name must be text") - where = f"step {index} ({name})" + if raw.get("name"): + where = f"step {index} ({name})" request = raw.get("request") if not isinstance(request, dict): @@ -178,9 +189,9 @@ def _parse_step(raw: Any, index: int) -> Step: if method not in METHODS: raise JourneyError(f"{where}: method {method} isn't supported") path = request.get("path") - if not isinstance(path, str) or not path.startswith("/"): - raise JourneyError(f"{where}: path must start with / (journeys only call the app " - "they're run against, never another site)") + if not isinstance(path, str) or not path.startswith("/") or path.startswith(("//", "/\\")): + raise JourneyError(f"{where}: path must start with a single / (it's relative to the " + "URL the journey runs against, and never reaches another site)") headers = request.get("headers") or {} if not isinstance(headers, dict) or not all(isinstance(v, str) for v in headers.values()): raise JourneyError(f"{where}: headers must map names to text") @@ -189,7 +200,9 @@ def _parse_step(raw: Any, index: int) -> Step: body = request.get("body") if body is not None and not isinstance(body, str): raise JourneyError(f"{where}: body must be text (use json for structured data)") - _check_secrets({"headers": headers, "json": request.get("json")}, where) + _check_secrets({"headers": headers, "json": request.get("json"), + "query": dict(parse_qsl(urlsplit(path).query)), + "body": _decode_body(body)}, where) return Step( name=name, method=method, path=path, @@ -247,6 +260,16 @@ def _only(data: dict, allowed: set[str], where: str) -> None: raise JourneyError(f"{where} has unknown field(s): {', '.join(map(str, unknown))}") +def _decode_body(body: str | None) -> Any: + """A string body as data, so secrets in it are checked too (JSON or a form).""" + if not body: + return None + try: + return json.loads(body) + except ValueError: + return dict(parse_qsl(body)) + + def _check_secrets(value: Any, where: str, key: str = "") -> None: """Refuse a literal value under a secret-looking key: secrets come from {{env.*}} (or an earlier step's extract), so they never end up committed in the file.""" @@ -256,8 +279,8 @@ def _check_secrets(value: Any, where: str, key: str = "") -> None: elif isinstance(value, list): for v in value: _check_secrets(v, where, key) - elif isinstance(value, str) and key and _SENSITIVE_KEY.search(key) \ - and not _PLACEHOLDER.search(value): + elif value is not None and key and _SENSITIVE_KEY.search(key) \ + and not (isinstance(value, str) and _SECRET_OK.fullmatch(value.strip())): raise JourneyError(f"{where}: {key} has a literal value. Put secrets in an " f"environment variable and write \"{{{{env.NAME}}}}\" instead") diff --git a/cli/src/prescale_cli/journeys/replay.py b/cli/src/prescale_cli/journeys/replay.py index b6e67ba..a5222e0 100644 --- a/cli/src/prescale_cli/journeys/replay.py +++ b/cli/src/prescale_cli/journeys/replay.py @@ -1,5 +1,9 @@ """Run a journey's steps against an app. +A step's path is relative to the URL the journey runs against: with +`http://host/app`, the path `/api/cart` is `http://host/app/api/cart`. Every +request, redirects included, must stay on that URL's scheme, host and port. + `check` replays a journey once, as a single user, to see whether it still works against the app as it is now. A step that fails its expectation here, with no load at all, means the journey is stale (the app changed), not that the app @@ -13,7 +17,6 @@ import time from dataclasses import dataclass, field from typing import Any -from urllib.parse import urljoin import httpx @@ -23,6 +26,26 @@ _SNIPPET = 160 +class OffOriginError(Exception): + """A request that would leave the app the journey runs against.""" + + +def step_url(base_url: str, path: str) -> str: + """The step's URL: PATH appended to BASE_URL, refused if it leaves its origin. + Appending (not urljoin) keeps a base path like /app and never lets a rendered + path such as //other-host/x change the host.""" + if not path.startswith("/") or path.startswith(("//", "/\\")): + raise OffOriginError(f"path {path!r} must start with a single /") + url = base_url.rstrip("/") + path + if _origin(httpx.URL(url)) != _origin(httpx.URL(base_url)): + raise OffOriginError(f"{url} isn't on {base_url}") + return url + + +def _origin(url: httpx.URL) -> tuple[str, str, int | None]: + return url.scheme, url.host, url.port + + @dataclass class StepCheck: step: Step @@ -59,8 +82,16 @@ async def check(journey: Journey, base_url: str, *, timeout: float = 10.0, env = dict(os.environ) if env is None else env result = JourneyCheck(journey) variables: dict[str, Any] = fresh_uniques() + origin = _origin(httpx.URL(base_url)) + + async def same_origin(request: httpx.Request) -> None: + # Runs for redirect hops too, so an app can't bounce a write elsewhere. + if _origin(request.url) != origin: + raise OffOriginError(f"redirected to {request.url.host}, outside the app") + async with httpx.AsyncClient(timeout=timeout, follow_redirects=True, - headers=client_headers(headers), transport=transport) as client: + headers=client_headers(headers), transport=transport, + event_hooks={"request": [same_origin]}) as client: for step in journey.steps: outcome = await run_step(client, step, base_url, variables, env) result.steps.append(outcome) @@ -75,18 +106,21 @@ async def run_step(client: httpx.AsyncClient, step: Step, base_url: str, into VARIABLES for the steps after it.""" outcome = StepCheck(step) try: - path = render(step.path, variables, env) - req_headers = render(step.headers, variables, env) + url = step_url(base_url, str(render(step.path, variables, env))) + req_headers = {k: str(v) for k, v in render(step.headers, variables, env).items()} payload = render(step.json, variables, env) if step.json is not None else None body = render(step.body, variables, env) if step.body is not None else None - except TemplateError as exc: + except (TemplateError, OffOriginError) as exc: outcome.problem = str(exc) return outcome start = time.perf_counter() try: - resp = await client.request(step.method, urljoin(base_url, path), - headers=req_headers, json=payload, content=body) + resp = await client.request(step.method, url, headers=req_headers, json=payload, + content=body) + except OffOriginError as exc: + outcome.problem = str(exc) + return outcome except httpx.TimeoutException: outcome.problem = "timed out" return outcome diff --git a/cli/tests/test_journeys.py b/cli/tests/test_journeys.py index 68c8142..889c3e0 100644 --- a/cli/tests/test_journeys.py +++ b/cli/tests/test_journeys.py @@ -79,7 +79,10 @@ def test_read_only_journey_does_not_write(): (_with(weight=0), "weight must be a positive number"), (_with(steps=[]), "steps must be a non-empty list"), (_with(extra=True), "unknown field(s): extra"), - (_step(0, request={"method": "POST", "path": "https://evil.example/x"}), "path must start with /"), + (_step(0, request={"method": "POST", "path": "https://evil.example/x"}), "single /"), + (_step(0, request={"method": "DELETE", "path": "//evil.example/x"}), "single /"), + (_with(weight=float("nan")), "weight must be a positive number"), + (_with(weight=float("inf")), "weight must be a positive number"), (_step(0, request={"method": "BREW", "path": "/"}), "method BREW isn't supported"), (_step(0, expect={"status": "ok"}), "HTTP status codes"), (_step(0, extract={"token": "cookie.sid"}), "body. or header."), @@ -120,6 +123,30 @@ def test_literal_secrets_are_refused(request_part): parse(doc) +@pytest.mark.parametrize("request_part", [ + {"body": '{"email": "a@b.co", "password": "hunter2"}'}, + {"body": "user=a&password=hunter2"}, + {"path": "/a?api_key=sk_live_123"}, + {"json": {"password": 123456}}, + {"headers": {"Authorization": "Basic dXNlcjpwdw== {{unique.id}}"}}, +]) +def test_literal_secrets_are_refused_anywhere(request_part): + request_part.setdefault("path", "/api/login") + doc = _step(0, request={"method": "POST", "path": "/api/login", **request_part}) + with pytest.raises(JourneyError, match="literal value"): + parse(doc) + + +@pytest.mark.parametrize("request_part", [ + {"json": {"passenger_name": "Ada", "passport": "X1", "token_type": "bearer", + "cookie_consent": "yes", "bypass": "no"}}, + {"headers": {"Authorization": "Bearer {{env.TOKEN}}", "X-Api-Key": "{{env.KEY}}"}}, + {"headers": {"Authorization": "Basic {{env.BASIC}}"}}, +]) +def test_ordinary_fields_and_placeholder_secrets_are_allowed(request_part): + parse(_step(0, request={"method": "POST", "path": "/api/login", **request_part})) + + def test_load_dir_reports_bad_files_and_duplicate_names(tmp_path): (tmp_path / "a.yaml").write_text(json.dumps(CHECKOUT)) (tmp_path / "b.yml").write_text(json.dumps(CHECKOUT)) # same name @@ -213,6 +240,68 @@ def handler(request): assert "couldn't find body.token" in result.failed.problem +def _one_step(path="/next", **request): + return parse({"version": 1, "name": "j", "steps": [ + {"request": {"method": "GET", "path": "/start"}, "extract": {"next": "body.next"}}, + {"request": {"method": "POST", "path": path, **request}}]}) + + +def test_rendered_path_cannot_change_host(): + seen = [] + + def handler(request): + seen.append(str(request.url)) + if request.url.path == "/start": + return httpx.Response(200, json={"next": "/evil.example/x", "pid": 1}) + return httpx.Response(200) + + j = _one_step(path="/{{next}}") + result = asyncio.run(check(j, "http://shop.test", env={}, transport=httpx.MockTransport(handler))) + assert seen == ["http://shop.test/start"] # nothing sent off the app + assert not result.ok and "single /" in result.failed.problem + + +def test_base_url_path_prefix_is_kept(): + seen = [] + + def handler(request): + seen.append(request.url.path) + return httpx.Response(200, json={"next": "x"}) + + asyncio.run(check(_one_step(), "http://shop.test/app/", env={}, + transport=httpx.MockTransport(handler))) + assert seen == ["/app/start", "/app/next"] + + +def test_redirect_to_another_host_fails_the_step(): + seen = [] + + def handler(request): + seen.append(request.url.host) + if request.url.path == "/start": + return httpx.Response(200, json={"next": "x"}) + return httpx.Response(307, headers={"Location": "https://evil.example/steal"}) + + result = asyncio.run(check(_one_step(), "http://shop.test", env={}, + transport=httpx.MockTransport(handler))) + assert "evil.example" not in seen + assert not result.ok and "outside the app" in result.failed.problem + + +def test_extracted_number_in_a_header_is_sent_as_text(): + sent = {} + + def handler(request): + if request.url.path == "/start": + return httpx.Response(200, json={"next": 7}) + sent["x-id"] = request.headers["x-id"] + return httpx.Response(200) + + result = asyncio.run(check(_one_step(headers={"X-Id": "{{next}}"}), "http://shop.test", + env={}, transport=httpx.MockTransport(handler))) + assert result.ok and sent["x-id"] == "7" + + # --- the command ------------------------------------------------------------- def _out(res): @@ -252,6 +341,15 @@ def test_verify_reports_invalid_files(tmp_path): assert "name must be lowercase" in _out(res) +def test_verify_prints_markup_like_names_safely(tmp_path): + doc = {"version": 1, "name": "j", "steps": [ + {"name": "[/bold] oops", "request": {"method": "GET", "path": "/"}}]} + res = CliRunner().invoke(cli, ["journeys", "verify", "http://127.0.0.1:9", + "--dir", str(_write(tmp_path, doc))]) + assert res.exit_code == 1, res.output + assert "[/bold] oops" in res.output + + def test_verify_unknown_journey_name(tmp_path): res = CliRunner().invoke(cli, ["journeys", "verify", "http://localhost:1", "--dir", str(_write(tmp_path)), "--journey", "nope"], diff --git a/examples/fragile-shop.py b/examples/fragile-shop.py index 2a6bd2d..c176011 100644 --- a/examples/fragile-shop.py +++ b/examples/fragile-shop.py @@ -59,8 +59,12 @@ async def _handle(reader, writer, pool): break name, _, value = header.decode(errors="replace").partition(":") headers[name.strip().lower()] = value.strip() - length = int(headers.get("content-length") or 0) - body = await reader.readexactly(length) if length else b"" + length = headers.get("content-length") or "0" + if not length.isdigit(): + writer.write(_resp(400, b"bad Content-Length")) + await writer.drain() + return + body = await reader.readexactly(int(length)) if int(length) else b"" if path.startswith("/api/"): out = await _api(method, path, headers, body, pool) @@ -111,7 +115,9 @@ async def _api(method, path, headers, body, pool): try: creds = json.loads(body or b"{}") except ValueError: - return _json(400, {"error": "send JSON"}) + creds = None + if not isinstance(creds, dict): + return _json(400, {"error": "send a JSON object"}) if not creds.get("email") or not creds.get("password"): return _json(401, {"error": "email and password required"}) await asyncio.sleep(0.02) # password hashing