diff --git a/AGENTS.md b/AGENTS.md index 05d123e..4608cae 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -151,7 +151,7 @@ for the http API surface, `tests/server/per_customer/` IS the per_customer stack - `bash scripts/deploy.sh image` — the agent-container deploy: build → push (auto-incremented immutable `vNN` tag) → CLI-update a gerp's runtime to the new digest, from the copy ECR replication makes in the gerp's own region, once it has arrived (FULL config carried — UpdateAgentRuntime replaces wholesale) → wait READY → re-pin the named endpoint (DEFAULT auto-tracks). The gerp is gradienterp by default (the dogfood takes an image first), `--gerp ` for another, `--all` for every active gerp — a fleet push is said, never implied; `--no-build` moves the named gerps onto the image already at the top of ECR, or `--no-build --tag vNN` onto that one. No terraform: the runtime tf reads `data.aws_ecr_image` `most_recent`, so a post-deploy plan is ALREADY clean (AgentCore exposes its config, unlike lambda code). No `agent_image_tag` var exists anymore; no fake gateway-target churn — routine image deploys change zero tf - `bash scripts/docker.sh --build|--run|--stop|--push ` — the underlying agent-container build/run primitives (`deploy.sh image` drives build/push; use directly for local container smoke) - `bash scripts/local-dev.sh --install` — once per clone: creates `.venv` (python3.12), installs the tests', the agent container's and the local servers' python deps, and runs `npm ci` in each Node lambda. Every script that runs `.venv/bin/python` (`test.sh`, `deploy.sh`, `investigate.sh`) needs it -- `bash scripts/local-dev.sh --start|--status|--stop` — the local dev stack as plain processes: moto :5000, the owner-app BFF :3000 (sign in at `/dev/login?sub=local-dev`; `/dev` lists the rest), the per_customer stack :8080, the Stripe stand-in :4242, the Cognito stand-in :4243, openlyoperated.biz :3001, and the pump (no port) +- `bash scripts/local-dev.sh --start|--status|--stop` — the local dev stack as plain processes: moto :5000, the owner-app BFF :3000 (sign in at `/dev/login?sub=local-dev`; `/dev` lists the rest), the per_customer stack :8080, the Stripe stand-in :4242, the Cognito stand-in :4243 (it signs every local token), openlyoperated.biz :3001, the pump (no port), the agent :8090 (the container entrypoint on `ANTHROPIC_API_KEY`), the web chat :8082 and the storage portal :8083 (the Node lambdas, `tests/server/node_lambda.mjs`). No AWS credentials anywhere in it - `bash scripts/e2e.sh [--env local|prod] [--suite smoke|full|lifecycle] [--configure]` — the browser suite (`tests/e2e`); local by default, starting the local stack when a surface is down; `--configure` installs the suite and writes the local seed (`LOCAL_GERPS`, a local owner of gradienterp) into `.env` - reading the fleet — open alarm tasks, a day's errors across every gerp, the agent's tokens, this month's cost per gerp: the commands are in `prod/platform/operator/AGENTS.md` § reading the fleet - `python3 tests/server/per_customer/snapshot.py` — re-take the per_customer manifest (routes + every function's env) from the running stack diff --git a/modules/agent/AGENTS.md b/modules/agent/AGENTS.md index 23a1231..481a3a7 100644 --- a/modules/agent/AGENTS.md +++ b/modules/agent/AGENTS.md @@ -87,7 +87,7 @@ Triggers / key outputs: ## layout -- `dev/` — local dev harness (no bedrock, no aws): stdin/stdout chat loop + importlib tool dispatch into sibling lambdas; `run.sh` wires `LOCAL_*` env under `out/agent-dev/`. +- `dev/` — the agent's turn, locally (no bedrock, no aws): `agent.py` (`run_turn`, Claude through the Anthropic SDK) + `tools.py` (the tools discovered from every `schema.json`, each call dispatched to the real handler — a Python one in-process, a Node one through `tests/server/node_lambda.mjs`). The container's `AnthropicEngine`, the local agent process and the tests import them. - `prompts/` — `bookkeeper.md` (the owner-facing persona; a new gerp's first conversation is the onboarding walk, a guide it retrieves — `onboarding/kb.md`), `broker.md` (the hub), `_shared/integrations.md` (the guides directive — setup AND workflow mechanics — composed into owner-facing modes at boot). - `docker/` — the AgentCore Runtime container (arm64; FastAPI `entrypoint.py` with the pluggable InferenceEngine + in-process tools; `registries/` bundles field registries for offline validation). - `lambdas/` — `chat/` (Node web-chat lambda + SPA) and `email/` (SES-inbound Python lambda). @@ -97,16 +97,26 @@ Tests live under `tests/agent/local/` and `tests/agent/integ/` — same conventi ## developing locally -No AWS needed. Three surfaces, same anthropic SDK + importlib tool dispatch under the hood. +No AWS needed. Three ways in, the same anthropic SDK + importlib tool dispatch under the hood. -**Interactive dev harness** — stdin/stdout loop, fastest iteration: +**The local stack** — the agent as a process, and the web chat in front of it: ``` export ANTHROPIC_API_KEY=sk-ant-... -bash modules/agent/dev/run.sh +bash scripts/local-dev.sh --start ``` -First run creates `.venv/` at the repo root and installs `anthropic` into it; subsequent runs skip bootstrap. +`entrypoint.py` runs under uvicorn on :8090 (`AnthropicEngine`, tools against moto), the chat lambda +on :8082 under `tests/server/node_lambda.mjs` with its SDK pointed at both. Sign in at +`localhost:3000/dev/login?sub=`, open the gerp, its chat link. Or curl the agent: + +``` +curl -sS -X POST localhost:8090/invocations -H 'content-type: application/json' \ + -H 'x-amzn-bedrock-agentcore-runtime-session-id: dev1' -d '{"prompt": "I sold a latte for $5.50 via Stripe"}' +``` + +Without the key the stack still starts; a turn answers with the error. `tests/server/AGENTS.md` has +the wiring. **Scripted smoke catalog**: @@ -145,7 +155,8 @@ The **engine owns the agent's authoritative session** — full message history p Local-only shortcuts: -- **`-v $(pwd):/repo`** — required by `AnthropicEngine`; entrypoint imports `dev/agent.py` and `dev/tools.py` from the mount so the prod image stays small. `StrandsEngine` doesn't need it. +- **`-v $(pwd):/repo`** — what `AnthropicEngine` reads `dev/agent.py` and `dev/tools.py` from inside the container, so the prod image stays small; outside it (the local stack) the entrypoint finds `dev/` beside itself. `StrandsEngine` needs neither. +- **`POST /runtimes//invocations`** — the path the SDK sends `InvokeAgentRuntime` to, answered the same as `/invocations`, so the local stack can point the chat lambda's SDK at this process (`AWS_ENDPOINT_URL_BEDROCK_AGENTCORE`). - **No Bedrock / no real Gateway** during local smoke — the Strands path constructs cleanly but the first MCP `list_tools_sync()` would fail without a deployed Gateway. The local bar is "engine selector flips and construction succeeds." ## deploying diff --git a/modules/agent/dev/agent.py b/modules/agent/dev/agent.py index 69c3f88..bf3dd0f 100644 --- a/modules/agent/dev/agent.py +++ b/modules/agent/dev/agent.py @@ -1,17 +1,12 @@ -"""Local dev harness for the gradientERP bookkeeper agent. +"""The agent's turn, locally: Claude through the Anthropic SDK, each tool call dispatched to the +real lambda handler (tools.py) — same data stores, same validation, no AWS. -Runs a stdin/stdout conversation loop backed by Claude via the Anthropic SDK. -Each tool call dispatches to a real accounting lambda handler via importlib — -same data stores, same validation, no AWS. Use this to iterate on the prompt, -tool schemas, and conversation flow before deploying to Bedrock AgentCore. +`run_turn` is what runs it. The container entrypoint's AnthropicEngine calls it for every turn of +the local agent process (`bash scripts/local-dev.sh --start`, :8090), and so do +tests/agent/local/smoke.py and test_bookkeeper_flow.py. -Prerequisites: export ANTHROPIC_API_KEY=sk-ant-... # the default # or: AGENT_PROVIDER=bedrock, which uses your AWS creds and production's model id - -Usage: - bash modules/agent/dev/run.sh - (handles the .venv bootstrap; then type; Ctrl-D or Ctrl-C to exit) """ import json @@ -93,35 +88,3 @@ def make_client(): raise SystemExit(f"unknown AGENT_PROVIDER={PROVIDER!r} — expected anthropic | bedrock") from anthropic import Anthropic return Anthropic() - - -def main(): - try: - client = make_client() - except ImportError: - sys.stderr.write("requires: pip3 install anthropic (or python3 -m pip install anthropic)\n") - sys.exit(1) - system = load_system_prompt() - messages = [] - - sys.stderr.write(f"agent ready — {PROVIDER}/{MODEL}, {len(TOOLS)} tools, business={BUSINESS_NAME}\n") - sys.stderr.write("type a message, Ctrl-D to exit\n\n") - - while True: - try: - user_input = input("> ") - except (EOFError, KeyboardInterrupt): - print() - return - - if not user_input.strip(): - continue - - messages.append({"role": "user", "content": user_input}) - reply = run_turn(client, system, messages) - print(reply) - print() - - -if __name__ == "__main__": - main() diff --git a/modules/agent/dev/run.sh b/modules/agent/dev/run.sh deleted file mode 100644 index 6da6189..0000000 --- a/modules/agent/dev/run.sh +++ /dev/null @@ -1,43 +0,0 @@ -#!/usr/bin/env bash -# Launch the local dev agent against a scratch dir under out/agent-dev/. -# Bootstraps a project-local venv at .venv/ on first run. -# Requires: export ANTHROPIC_API_KEY=... -# -# Usage: -# bash modules/agent/dev/run.sh -# BUSINESS_NAME="Maria's Cafe" bash modules/agent/dev/run.sh -# AGENT_MODEL=claude-opus-4-7 bash modules/agent/dev/run.sh - -set -euo pipefail - -REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../../.." && pwd)" -VENV="$REPO_ROOT/.venv" -SCRATCH="$REPO_ROOT/out/agent-dev" -LOGS="$REPO_ROOT/logs/agent-dev" - -# bootstrap venv + deps on first run -if [[ ! -d "$VENV" ]]; then - echo "creating venv at $VENV" >&2 - python3 -m venv "$VENV" -fi -# shellcheck disable=SC1091 -source "$VENV/bin/activate" - -if ! python -c "import anthropic" 2>/dev/null; then - echo "installing anthropic into $VENV" >&2 - pip install --quiet anthropic -fi - -mkdir -p "$SCRATCH" "$LOGS" - -export LOCAL_LEDGER="$SCRATCH/ledger.jsonl" -export LOCAL_PENDING="$SCRATCH/pending.jsonl" -export LOCAL_BALANCES="$SCRATCH/balances.jsonl" -export LOCAL_CLASSIFICATIONS="$SCRATCH/classifications.jsonl" -export LOCAL_S3="$SCRATCH/reports" -export LOCAL_LOGS="$LOGS" - -# ensure lambdas take the local path -unset AWS_LAMBDA_FUNCTION_NAME - -python "$REPO_ROOT/modules/agent/dev/agent.py" diff --git a/modules/agent/dev/tools.py b/modules/agent/dev/tools.py index 484a0d1..625ee43 100644 --- a/modules/agent/dev/tools.py +++ b/modules/agent/dev/tools.py @@ -67,7 +67,7 @@ def _discover(): out = {} for schema in sorted(REPO_ROOT.glob("modules/*/lambdas/*/schema.json")): src = schema.parent - if not (src / "main.py").exists(): + if not ((src / "main.py").exists() or (src / "index.mjs").exists()): continue out[src.name] = (schema, src) return out @@ -95,6 +95,8 @@ def invoke(name, tool_input): if not entry: return {"error": f"unknown tool '{name}'", "known": sorted(_CATALOG)[:20]} _, src = entry + if not (src / "main.py").exists(): + return _invoke_node(src.name, tool_input) from aws import _bundle_dirs import importlib.util main = src / "main.py" @@ -106,12 +108,30 @@ def invoke(name, tool_input): module_spec.loader.exec_module(mod) response = mod.handler(tool_input, None) + return _unwrap(response) + + +def _unwrap(response): if isinstance(response, dict) and "body" in response: body = response["body"] return json.loads(body) if isinstance(body, str) else body return response +def _invoke_node(name, tool_input): + """A Node tool (an `index.mjs`) runs in the local Node process, tests/server/node_lambda.mjs, + started by scripts/local-dev.sh at LOCAL_NODE_URL: the event in, the handler's return out.""" + import urllib.request + base = os.environ.get("LOCAL_NODE_URL", "http://127.0.0.1:8082").rstrip("/") + req = urllib.request.Request(f"{base}/invoke/{name}", data=json.dumps(tool_input).encode(), + headers={"content-type": "application/json"}, method="POST") + try: + with urllib.request.urlopen(req, timeout=120) as r: + return _unwrap(json.loads(r.read() or b"null")) + except OSError as e: + return {"error": f"the local Node process at {base} did not answer: {e}"} + + if __name__ == "__main__": print(f"{len(TOOLS)} tools · {_ENV_VARS} env vars") for t in TOOLS: diff --git a/modules/agent/docker/entrypoint.py b/modules/agent/docker/entrypoint.py index ec0e46d..405acc3 100644 --- a/modules/agent/docker/entrypoint.py +++ b/modules/agent/docker/entrypoint.py @@ -327,27 +327,45 @@ class AnthropicEngine(InferenceEngine): """ def __init__(self) -> None: - dev_path = Path("/repo/modules/agent/dev") + dev_path = Path("/repo/modules/agent/dev") # the container, the repo mounted (scripts/docker.sh --run) + if not dev_path.is_dir(): + dev_path = Path(__file__).resolve().parents[1] / "dev" # a process in the repo (scripts/local-dev.sh) if not dev_path.is_dir(): raise RuntimeError( - f"AnthropicEngine expects the dev harness mounted at {dev_path}. " + f"AnthropicEngine expects the dev harness at {dev_path}. " "run with `bash scripts/docker.sh --run` (mounts $PWD:/repo) — " "or set BEDROCK_MODEL_ID + GATEWAY_URL to select StrandsEngine instead." ) sys.path.insert(0, str(dev_path)) - from anthropic import Anthropic from agent import run_turn as _dev_run_turn # dev/agent.py - self._client = Anthropic() + self._client = None # made on the first turn: the process starts without ANTHROPIC_API_KEY, a turn needs it self._dev_run_turn = _dev_run_turn # local smoke has no Strands SessionManager, so the engine keeps its own # volatile restore store (per-container; matches the dev harness lifetime). self._store = InMemoryStore() + @staticmethod + def _system(blocks): + """`_system_blocks` is in Bedrock's shape. The Anthropic API wants `{"type": "text", ...}` + blocks, and a cache point is `cache_control` on the block before it.""" + if isinstance(blocks, str): + return blocks + out = [] + for b in blocks: + if "text" in b: + out.append({"type": "text", "text": b["text"]}) + elif "cachePoint" in b and out: + out[-1]["cache_control"] = {"type": "ephemeral"} + return out + def run_turn(self, system, session_id, user_message): + if self._client is None: + from anthropic import Anthropic + self._client = Anthropic() prior = self._store.load(session_id) messages = list(prior) messages.append({"role": "user", "content": user_message}) - reply = self._dev_run_turn(self._client, system, messages) + reply = self._dev_run_turn(self._client, self._system(system), messages) new_messages = messages[len(prior):] self._store.append(session_id, new_messages) return reply, new_messages @@ -2613,6 +2631,14 @@ async def invoke(payload: dict, request: Request): return {"error": f"{type(e).__name__}: {e}", "tb": traceback.format_exc().splitlines()[-10:]} +@app.post("/runtimes/{arn:path}/invocations") +async def invoke_runtime(arn: str, payload: dict, request: Request): + """The path the SDK sends InvokeAgentRuntime to. The local stack points the web chat lambda's + SDK at this process (AWS_ENDPOINT_URL_BEDROCK_AGENTCORE), so its call lands here unchanged; + the arn and the qualifier are not read.""" + return await invoke(payload, request) + + async def _invoke_inner(payload: dict, request: Request) -> dict: # AgentCore HTTP protocol contract: body is {"prompt": "..."}. Local smoke # uses the same shape for consistency — no dual parsing. diff --git a/modules/agent/lambdas/chat/index.mjs b/modules/agent/lambdas/chat/index.mjs index c31cc9a..43e34d2 100644 --- a/modules/agent/lambdas/chat/index.mjs +++ b/modules/agent/lambdas/chat/index.mjs @@ -43,7 +43,8 @@ const MEMORY_ACTOR_ID = process.env.MEMORY_ACTOR_ID || "agent-session"; // must const SESSIONS_BUCKET = process.env.SESSIONS_BUCKET || ""; // S3SessionManager bucket — the agent's authoritative session + checkpoint const SESSIONS_PREFIX = process.env.SESSIONS_PREFIX || "agent-sessions/"; // must match the container's SESSIONS_PREFIX default const BUSINESS_NAME = process.env.BUSINESS_NAME || ""; // the gerp's business name — header shows it instead of the login email -const ISSUER = POOL ? `https://cognito-idp.${REGION}.amazonaws.com/${POOL}` : ""; +// the pool's issuer; COGNITO_ISSUER names another (the local stand-in, tests/server/cognito) +const ISSUER = process.env.COGNITO_ISSUER || (POOL ? `https://cognito-idp.${REGION}.amazonaws.com/${POOL}` : ""); const CLOCK_LEEWAY = 60; // invoke_agent_runtime wants the RUNTIME arn + the endpoint name as `qualifier`, not diff --git a/scripts/local-dev.sh b/scripts/local-dev.sh index fbf883f..983850b 100644 --- a/scripts/local-dev.sh +++ b/scripts/local-dev.sh @@ -16,6 +16,9 @@ # :4243 cognito tests/server/cognito/server.py — signup, the login-email change, refresh; no Cognito # :3001 oob.biz prod/openlyoperated_biz/web — the public dashboard, reading the live api # pump tests/server/pump.py — the streams and queues moto doesn't deliver; no port +# :8090 agent modules/agent/docker/entrypoint.py — the agent on ANTHROPIC_API_KEY, its tools in-process; no Bedrock +# :8082 chat tests/server/node_lambda.mjs — the web chat Function URL (modules/agent/lambdas/chat); /invoke for the Node tools +# :8083 portal tests/server/node_lambda.mjs — the storage portal Function URL (modules/storage/lambdas/ui) # # Plain processes, not containers: all three are Python, so on macOS a container would put a VM and # a filesystem layer inside the edit→see-it loop, and mounting /repo to get live edits back only @@ -29,7 +32,8 @@ # Against a gerp's live gateway with a real login (LOCAL_GATEWAY_URL unset), the seed # is the dogfood's: the SecureString /gradienterp/customers/gradienterp/local_dev/LOCAL_GERPS # in the gradienterp gerp's account, also the LOCAL_GERPS repo secret for a cloud run -# MOTO_PORT / BFF_PORT / GW_PORT to move a surface off a busy port +# ANTHROPIC_API_KEY the agent's model key. Without it the stack starts and a chat turn answers with the error +# MOTO_PORT / BFF_PORT / GW_PORT / AGENT_PORT / CHAT_PORT / PORTAL_PORT to move a surface off a busy port set -euo pipefail @@ -47,6 +51,9 @@ GW_PORT="${GW_PORT:-8080}" STRIPE_PORT="${STRIPE_PORT:-4242}" COGNITO_PORT="${COGNITO_PORT:-4243}" BIZ_PORT="${BIZ_PORT:-3001}" +AGENT_PORT="${AGENT_PORT:-8090}" +CHAT_PORT="${CHAT_PORT:-8082}" +PORTAL_PORT="${PORTAL_PORT:-8083}" # the bff forwards a gerp-scoped request to its gateway; locally that is the per_customer image. # Unset to forward at the real deployed gateway instead (needs a real Hosted-UI token). export LOCAL_GATEWAY_URL="${LOCAL_GATEWAY_URL-http://localhost:$GW_PORT}" @@ -66,6 +73,15 @@ export PYTHONPATH="$REPO_ROOT/modules:$REPO_ROOT/modules/clock:$REPO_ROOT/module # event-emitting lambdas would start putting to the PRODUCTION bus. unset AWS_LAMBDA_FUNCTION_NAME export LOCAL_AWS_ENDPOINT="${LOCAL_AWS_ENDPOINT:-http://localhost:$MOTO_PORT}" +# the Cognito stand-in signs every local token; the bff image and the chat lambda verify against its jwks +export LOCAL_COGNITO_URL="${LOCAL_COGNITO_URL-http://127.0.0.1:$COGNITO_PORT}" +# the gerp row's chat_url: the chat Function URL, served by the Node process below +export LOCAL_CHAT_URL="${LOCAL_CHAT_URL-http://localhost:$CHAT_PORT/}" +# where dev/tools.py and the image send a Node function's event +export LOCAL_NODE_URL="http://127.0.0.1:$CHAT_PORT" +# the Node lambdas' SDK clients: moto, and the agent process for InvokeAgentRuntime. The chat +# lambda's issuer is the stand-in and its Memory mirror is off — moto has no Memory data plane +NODE_ENV="env AWS_REGION=us-east-1 AWS_ACCESS_KEY_ID=local AWS_SECRET_ACCESS_KEY=local AWS_ENDPOINT_URL=$LOCAL_AWS_ENDPOINT AWS_ENDPOINT_URL_BEDROCK_AGENTCORE=http://127.0.0.1:$AGENT_PORT COGNITO_ISSUER=$LOCAL_COGNITO_URL COGNITO_CLIENT_ID=local COGNITO_DOMAIN_PREFIX= MEMORY_ID=" # name|port|signature|start-command. The signature is what `pgrep -f` finds a running copy by, # however it was started: the module or script the surface runs. @@ -78,6 +94,9 @@ biz|$BIZ_PORT|tests.server.openlyoperated_biz.server:app|$PY -m uvicorn tests.se stripe|$STRIPE_PORT|tests.server.stripe.server:app|$PY -m uvicorn tests.server.stripe.server:app --host 127.0.0.1 --port $STRIPE_PORT cognito|$COGNITO_PORT|tests.server.cognito.server:app|$PY -m uvicorn tests.server.cognito.server:app --host 127.0.0.1 --port $COGNITO_PORT pump|-|tests/server/pump.py|$PY $REPO_ROOT/tests/server/pump.py +agent|$AGENT_PORT|modules/agent/docker|$PY -m uvicorn entrypoint:app --app-dir $REPO_ROOT/modules/agent/docker --host 127.0.0.1 --port $AGENT_PORT +chat|$CHAT_PORT|node_lambda.mjs --port $CHAT_PORT|$NODE_ENV node $REPO_ROOT/tests/server/node_lambda.mjs --port $CHAT_PORT --front modules/agent/lambdas/chat +portal|$PORTAL_PORT|node_lambda.mjs --port $PORTAL_PORT|$NODE_ENV node $REPO_ROOT/tests/server/node_lambda.mjs --port $PORTAL_PORT --front modules/storage/lambdas/ui EOF } @@ -209,7 +228,9 @@ case "${1:---status}" in echo " local AWS http://localhost:$MOTO_PORT" echo " owner app http://localhost:$BFF_PORT (a registered Cognito callback origin)" echo " gateway http://localhost:$GW_PORT /webhooks/ - oob.biz http://localhost:$BIZ_PORT the public dashboard, on the live api" + oob.biz http://localhost:$BIZ_PORT the public dashboard, on the live api + agent http://localhost:$AGENT_PORT POST /invocations {\"prompt\": …}, on ANTHROPIC_API_KEY$([[ -n "${ANTHROPIC_API_KEY:-}" ]] || echo " (unset)") + chat http://localhost:$CHAT_PORT the web chat: sign in at the owner app, open a gerp, its chat link" ;; --stop) diff --git a/tests/AGENTS.md b/tests/AGENTS.md index 4ddfaab..104823c 100644 --- a/tests/AGENTS.md +++ b/tests/AGENTS.md @@ -244,8 +244,11 @@ tests/ ## the local dev stack -`bash scripts/local-dev.sh --start` brings up moto plus the three stacks as plain processes; -`--restart` cycles them, `--install` does the one-time pip step. +`bash scripts/local-dev.sh --start` brings up moto plus the three stacks as plain processes, and +with them the agent (:8090, the container entrypoint on `ANTHROPIC_API_KEY`), the web chat (:8082) +and the storage portal (:8083), the Node lambdas run by `tests/server/node_lambda.mjs`; `--restart` +cycles them, `--install` does the one-time pip step. Sign in at +`localhost:3000/dev/login?sub=`, open the gerp, and its chat link reaches the agent. ``` curl -X POST localhost:8080/webhooks/stripe -d @tests/testdata/stripe/charge.succeeded.json @@ -359,7 +362,9 @@ top-to-bottom as "what the ingestion pipeline does" — that is its job. Secrets Manager, STS, Lambda, Cognito, Scheduler and Textract. It answers AgentCore's CONTROL plane but not its data plane (`invoke_agent_runtime` 404s), and it has no `location`. For AgentCore that matters less than it sounds: every fault there has been in the ARGUMENTS, which a - fake captures better than a live runtime would — see `test_poke_arn_split.py`. + fake captures better than a live runtime would — see `test_poke_arn_split.py`. The web chat's + invoke is the exception: the local agent process answers the SDK's path, and + `tests/agent/local/test_local_chat.py` runs the chat lambda against it. - **The third parties on the other side of an outbound HTTP call** — Plaid, and Stripe / Square / PayPal's own APIs. Each is stubbed with a hand-written lambda per test, so what is pinned is OUR side of the conversation and nothing else. `tests/TODO.md` § 1 has the seam-by-seam list and what diff --git a/tests/agent/local/test_local_agent_process.py b/tests/agent/local/test_local_agent_process.py new file mode 100644 index 0000000..4b9898b --- /dev/null +++ b/tests/agent/local/test_local_agent_process.py @@ -0,0 +1,64 @@ +"""The container entrypoint as a local process (`scripts/local-dev.sh`, :8090): it answers the path +the SDK sends `InvokeAgentRuntime` to, its Anthropic engine finds the dev harness beside itself +with no `/repo` mount and starts without a key, and it hands the Anthropic API the system blocks +in that API's shape.""" + +import importlib.util +import os +import sys +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[3] + + +def _load(): + os.environ.setdefault("AGENT_MODE", "bookkeeper") + os.environ.setdefault("AWS_DEFAULT_REGION", "us-east-1") + os.environ.setdefault("BUSINESS_NAME", "Test Co") + os.environ.setdefault("CUSTOMER_ID", "testco") + os.environ.setdefault("BEDROCK_MODEL_ID", "us.anthropic.claude-sonnet-4-6") # StrandsEngine, lazy + os.environ.setdefault("GATEWAY_URL", "https://example.invalid/mcp") + path = REPO_ROOT / "modules/agent/docker/entrypoint.py" + spec = importlib.util.spec_from_file_location("agent_entrypoint_local", path) + mod = importlib.util.module_from_spec(spec) + sys.modules["agent_entrypoint_local"] = mod + spec.loader.exec_module(mod) + return mod + + +def test_the_sdks_invoke_path_is_answered_beside_invocations(): + mod = _load() + paths = {getattr(r, "path", None) for r in mod.app.routes} + assert "/invocations" in paths + assert "/runtimes/{arn:path}/invocations" in paths, "where the SDK sends InvokeAgentRuntime" + + +def test_the_system_blocks_reach_the_anthropic_api_in_its_shape(): + mod = _load() + blocks = mod._system_blocks("the persona", "## today\n\nMonday") + assert mod.AnthropicEngine._system(blocks) == [ + {"type": "text", "text": "the persona", "cache_control": {"type": "ephemeral"}}, + {"type": "text", "text": "## today\n\nMonday"}, + ], "a text block gets its type; the cache point becomes cache_control on the block before it" + assert mod.AnthropicEngine._system("plain") == "plain" + + +def test_the_anthropic_engine_finds_the_dev_harness_in_the_repo_and_starts_without_a_key(): + mod = _load() + key = os.environ.pop("ANTHROPIC_API_KEY", None) + try: + assert not Path("/repo").exists(), "this test runs in the checkout, not the container" + engine = mod.AnthropicEngine() + assert engine._client is None, "the client is made on the first turn, so the process starts with no key" + assert callable(engine._dev_run_turn) + finally: + if key is not None: + os.environ["ANTHROPIC_API_KEY"] = key + + +if __name__ == "__main__": + for _name, _fn in sorted(globals().items()): + if _name.startswith("test_") and callable(_fn): + _fn() + print(f"ok {_name}") + print("all local agent process tests passed") diff --git a/tests/agent/local/test_local_chat.py b/tests/agent/local/test_local_chat.py new file mode 100644 index 0000000..a821064 --- /dev/null +++ b/tests/agent/local/test_local_chat.py @@ -0,0 +1,159 @@ +"""The web chat door, locally: `modules/agent/lambdas/chat/index.mjs` run by `tests/server/node_lambda.mjs` +with its SDK clients pointed by env at moto and at an agent process, its issuer the Cognito stand-in. + +The contracts this pins, each one a thing a side assumes of the other: + - the chat lambda verifies a bearer against the issuer `COGNITO_ISSUER` names, so the stand-in's + signed token gets in and an unsigned one is refused + - `AWS_ENDPOINT_URL_BEDROCK_AGENTCORE` lands `InvokeAgentRuntime` on `POST /runtimes//invocations` + with the session id in its header and the turn's payload as its body — the path the agent process + answers — and a buffered `{"response"}` comes back to the browser as one text chunk + - the chats row the lambda keeps is written to the table on moto that `CHATS_TABLE` names + +No model: the agent is a stub answering a fixed reply, so the test costs nothing and runs anywhere.""" + +import json +import os +import socket +import subprocess +import sys +import threading +import time +import urllib.error +import urllib.request +from http.server import BaseHTTPRequestHandler, HTTPServer +from pathlib import Path + +REPO = Path(__file__).resolve().parents[3] +sys.path.insert(0, str(REPO / "tests")) +sys.path.insert(0, str(REPO / "modules" / "aws")) +os.environ.pop("AWS_LAMBDA_FUNCTION_NAME", None) +from helpers.localaws import make_table # noqa: E402 +sys.path.insert(0, str(REPO / "tests" / "server" / "cognito")) +import standin # noqa: E402 + +OWNER = "owner-sub-0001" +ARN = "arn:aws:bedrock-agentcore:us-east-1:000000000000:runtime/r1/runtime-endpoint/ep1" + + +def _free_port(): + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +def _wait(url, proc=None): + for _ in range(200): + try: + urllib.request.urlopen(url, timeout=1) + return + except urllib.error.HTTPError: + return + except (urllib.error.URLError, OSError): + if proc is not None and proc.poll() is not None: + raise RuntimeError(f"process exited: {proc.stdout.read()}") + time.sleep(0.05) + raise RuntimeError(f"{url} did not come up") + + +def _http(method, url, body=None, headers=None): + data = json.dumps(body).encode() if body is not None else None + req = urllib.request.Request(url, data=data, method=method, headers={"content-type": "application/json", **(headers or {})}) + try: + with urllib.request.urlopen(req, timeout=30) as r: + return r.status, r.read().decode() + except urllib.error.HTTPError as e: + return e.code, e.read().decode() + + +class _AgentStub(BaseHTTPRequestHandler): + """What the agent process answers: a buffered turn. Records what it was sent.""" + seen: list = [] + + def do_POST(self): + body = self.rfile.read(int(self.headers.get("content-length") or 0)) + _AgentStub.seen.append({"path": self.path, "headers": {k.lower(): v for k, v in self.headers.items()}, "body": json.loads(body or b"{}")}) + out = json.dumps({"response": "hi from the stub", "session_id": "x"}).encode() + self.send_response(200) + self.send_header("content-type", "application/json") + self.send_header("content-length", str(len(out))) + self.end_headers() + self.wfile.write(out) + + def log_message(self, *a): # quiet + pass + + +def test_the_chat_door_runs_locally_against_the_stand_in_and_an_agent_process(): + endpoint = os.environ.get("LOCAL_AWS_ENDPOINT", "http://localhost:5000") + chats = make_table("agent-chats") + from aws import client + client("ssm").put_parameter(Name="/local-chat-test/owner_sub", Value=OWNER, Type="String", Overwrite=True) + + stub = HTTPServer(("127.0.0.1", 0), _AgentStub) + threading.Thread(target=stub.serve_forever, daemon=True).start() + agent_url = f"http://127.0.0.1:{stub.server_port}" + + cognito, cog_url = standin.start() + chat_port = _free_port() + chat_url = f"http://127.0.0.1:{chat_port}" + node_env = { + "AWS_REGION": "us-east-1", "AWS_ACCESS_KEY_ID": "local", "AWS_SECRET_ACCESS_KEY": "local", + "AWS_ENDPOINT_URL": endpoint, "AWS_ENDPOINT_URL_BEDROCK_AGENTCORE": agent_url, + "COGNITO_ISSUER": cog_url, "COGNITO_CLIENT_ID": "local", "COGNITO_DOMAIN_PREFIX": "", + "MEMORY_ID": "", "SESSIONS_BUCKET": "", "CONTACTS_TABLE": "", + "CHATS_TABLE": chats, "OWNER_SUB_PARAM": "/local-chat-test/owner_sub", + "AGENT_RUNTIME_ENDPOINT_ARN": ARN, "BUSINESS_NAME": "Test Co", + } + node = subprocess.Popen(["node", str(REPO / "tests/server/node_lambda.mjs"), "--port", str(chat_port), + "--front", "modules/agent/lambdas/chat"], + env={**os.environ, **node_env}, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True) + try: + _wait(chat_url + "/", node) + token = lambda sub: json.loads(_http("GET", f"{cog_url}/token?sub={sub}&email={sub}@localhost")[1])["IdToken"] # noqa: E731 + + # the page, and the door's refusals: no token, an unsigned token, a signed token for a non-member + status, page = _http("GET", chat_url + "/") + assert status == 200 and " (process.env.E2E_ENV || "prod") === "local"; diff --git a/tests/e2e/helpers/login.mjs b/tests/e2e/helpers/login.mjs index ba393f1..f0e700c 100644 --- a/tests/e2e/helpers/login.mjs +++ b/tests/e2e/helpers/login.mjs @@ -1,12 +1,13 @@ import { expect, test } from "@playwright/test"; -import { isLocal } from "./env.mjs"; +import { isLocal, resolveEnv } from "./env.mjs"; -// An unsigned JWT. Nothing local verifies one: the SPA decodes the payload for display, and the bff -// image decodes the bearer's claims because production's APIGW authorizer — the thing that DOES -// verify — has no local equivalent. Reproducing a signature would only be reproducing AWS. -export function localIdToken({ sub, email }) { - const b64 = (o) => Buffer.from(JSON.stringify(o)).toString("base64url"); - return `${b64({ alg: "none", typ: "JWT" })}.${b64({ sub, email, token_use: "id" })}.local`; +// An id token the local Cognito stand-in (tests/server/cognito) signed for `sub`. The bff image and +// the chat lambda verify every bearer against the stand-in's jwks, so this is the one token that gets in. +export async function localIdToken({ sub, email }) { + const q = new URLSearchParams({ sub, email }); + const r = await fetch(`${resolveEnv().cognito}/token?${q}`); + if (!r.ok) throw new Error(`the Cognito stand-in did not issue a token (${r.status}) — is the local stack up?`); + return (await r.json()).IdToken; } // Drive the Cognito login. The SPA's "Log in" → Cognito Managed Login (email+password) → @@ -18,7 +19,7 @@ export async function login(page, { email, password, sub }) { // The SPA holds its session in sessionStorage and the PKCE callback is what puts it there. // Seed it before first paint and the app boots logged in, exercising every screen after the // login gate — which is the half a local run is for. - const token = localIdToken({ sub: sub || process.env.E2E_LOCAL_SUB || email, email }); + const token = await localIdToken({ sub: sub || process.env.E2E_LOCAL_SUB || email, email }); await page.addInitScript((t) => sessionStorage.setItem("id_token", t), token); await page.goto("/"); await expect(page.locator('[data-view="homeScreen"]')).toBeVisible({ timeout: 15_000 }); diff --git a/tests/gradienterp_cloud/local/test_dev_login.py b/tests/gradienterp_cloud/local/test_dev_login.py index 051ce1e..d6c501c 100644 --- a/tests/gradienterp_cloud/local/test_dev_login.py +++ b/tests/gradienterp_cloud/local/test_dev_login.py @@ -1,16 +1,23 @@ """The local stack's sign-in: `GET /dev/login?sub=` answers a page that stores a token for that -sub and goes to /. The route was once registered with its query string in the path, which no request -matches, so the catch-all served the app page and nobody was signed in.""" +sub and goes to /. The token is one the Cognito stand-in signed — the bff image and the chat lambda +accept no other — so the verb needs the stand-in up, and says so when it is not. The route was once +registered with its query string in the path, which no request matches, so the catch-all served the +app page and nobody was signed in.""" -import base64 import json +import os import re import sys +import urllib.request from pathlib import Path REPO = Path(__file__).resolve().parents[3] sys.path.insert(0, str(REPO)) sys.path.insert(0, str(REPO / "modules" / "aws")) +sys.path.insert(0, str(REPO / "tests" / "server")) +sys.path.insert(0, str(REPO / "tests" / "server" / "cognito")) +import _jwt # noqa: E402 +import standin # noqa: E402 from tests.server.bff import dev # noqa: E402 @@ -25,16 +32,36 @@ def test_the_login_url_the_docs_give_reaches_the_login_verb(): assert all("?" not in r.path for r in dev.router.routes), "a query string in a route path matches nothing" -def test_the_page_stores_a_token_for_the_sub_and_goes_home(): - body = dev.login(sub="local-dev").body.decode() - token = re.search(r'setItem\(\'id_token\', "([^"]+)"\)', body).group(1) - payload = json.loads(base64.urlsafe_b64decode(token.split(".")[1] + "==")) - assert payload["sub"] == "local-dev" and payload["email"] == "local-dev@localhost" - assert "location.replace('/')" in body +def test_the_page_stores_the_stand_ins_token_for_the_sub_and_goes_home(): + proc, url = standin.start() + try: + os.environ["LOCAL_COGNITO_URL"] = url + body = dev.login(sub="local-dev").body.decode() + token = re.search(r'setItem\(\'id_token\', "([^"]+)"\)', body).group(1) + with urllib.request.urlopen(f"{url}/.well-known/jwks.json", timeout=3) as r: + jwks = {k["kid"]: k for k in json.loads(r.read())["keys"]} + claims = _jwt.verify(token, jwks, url) + assert claims["sub"] == "local-dev" and claims["email"] == "local-dev@localhost" and claims["token_use"] == "id" + assert "location.replace('/')" in body + finally: + proc.terminate() + proc.wait(timeout=5) + os.environ.pop("LOCAL_COGNITO_URL", None) + + +def test_without_the_stand_in_the_verb_says_so(): + os.environ["LOCAL_COGNITO_URL"] = "http://127.0.0.1:1" + try: + r = dev.login(sub="local-dev") + assert r.status_code == 503 + assert "127.0.0.1:1" in r.body.decode() and "local-dev.sh --start" in r.body.decode() + finally: + os.environ.pop("LOCAL_COGNITO_URL", None) if __name__ == "__main__": - for _n in [k for k in dir() if k.startswith("test_")]: - globals()[_n]() - print(f"ok {_n}") + for _name, _fn in sorted(globals().items()): + if _name.startswith("test_") and callable(_fn): + _fn() + print(f"ok {_name}") print("all dev login tests passed") diff --git a/tests/server/AGENTS.md b/tests/server/AGENTS.md index 40f00a9..a047390 100644 --- a/tests/server/AGENTS.md +++ b/tests/server/AGENTS.md @@ -14,7 +14,8 @@ exist yet. ## current features - **three stacks as plain processes**, started together by `bash scripts/local-dev.sh --start`: - `bff` :3000 · `per_customer` :8080 · `openlyoperated_biz` :3001, over moto on :5000. `--stop` + `bff` :3000 · `per_customer` :8080 · `openlyoperated_biz` :3001, over moto on :5000, with the + agent on :8090, the web chat on :8082 and the storage portal on :8083 (below). `--stop` ends every running copy of each surface, found by the module it runs (`pgrep -f`) rather than by pidfile or port, so a copy another session started by hand goes too; `--start` does the same first and starts fresh. A non-surface process on a surface's port is reported and left alone. @@ -30,6 +31,21 @@ exist yet. - **`image.overlay.json`** for a route you are still designing, before it exists in AWS at all - **events are delivered**, not just accepted: `pump.py` runs the bus's real rules and hands each match to its target handler, so an emit reaches its consumer — and every emit is logged +- **the Node lambdas run too.** `node_lambda.mjs` is `_image.py` for an `index.mjs`: the snapshot's + env applied, the handler imported from source and re-imported on edit, the Function URL event + built from the request. One process per Function URL — the web chat (`modules/agent/lambdas/chat`) + on :8082, the storage portal (`modules/storage/lambdas/ui`) on :8083 — and `POST /invoke/` on + either for a Node tool (`manage_storage`, `inspect_document`), which `dev/tools.py` and the + image's `/fn/` mount call. The SDK inside is pointed by env: `AWS_ENDPOINT_URL` at moto, + `AWS_ENDPOINT_URL_BEDROCK_AGENTCORE` at the agent process. No handler carries a local branch +- **the agent is a process**, :8090 — the container's `entrypoint.py` under uvicorn, its + `AnthropicEngine` on `ANTHROPIC_API_KEY`, tools dispatched in-process to the real handlers against + this stack. It answers `POST /runtimes//invocations`, the path the SDK sends + `InvokeAgentRuntime` to, so the chat lambda's call lands on it unchanged. The stack starts without + the key; a turn then answers with the error +- **every local token is signed.** The Cognito stand-in (:4243) signs RS256 with a key it makes at + start and lists at `/.well-known/jwks.json`; the bff image and the chat lambda verify every bearer + against it. `/dev/login`, the e2e login helper and `GET /token?sub=&email=` all hand out its tokens ## the pieces @@ -37,7 +53,11 @@ exist yet. tests/server/ snapshot.py # → tests/server//image.json, by asking AWS _image.py # the shared runner: state, routes, dispatch, $default + node_lambda.mjs # the same for a Node lambda: a Function URL on its port, /invoke/ + _jwt.py # RS256 sign/verify — the stand-in signs, the image verifies pump.py # the bus's lambda targets, delivered (no port) + cognito/ # the Cognito stand-in :4243 + stripe/ # the Stripe stand-in :4242 bff/ # server.py + image.json :3000 per_customer/ # server.py + image.json :8080 openlyoperated_biz/ # server.py + image.json :3001 @@ -61,20 +81,31 @@ consequence is that a local failure is a real failure: a query naming an index t fails here, a rule pattern that matches nothing delivers nothing here, and a handler that reads an env var the deployment does not set raises here. -Three things are deliberately NOT reproduced, because reproducing them would only be reproducing -AWS: the JWT authorizer (claims are decoded, not verified), IAM (every call is allowed), and the -model behind the agent — a real network call either way, and the one thing no emulator should fake. +Two things are deliberately NOT reproduced, because reproducing them would only be reproducing +AWS: IAM (every call is allowed), and the model behind the agent — a real network call either way, +and the one thing no emulator should fake. The JWT authorizer's check IS here: claims come from a +bearer the Cognito stand-in signed, verified against its jwks, or from `x-debug-sub`. -**The agent, locally.** `modules/agent/dev` runs the turn loop in-process, with the 83 tools -DISCOVERED from their `schema.json` (the same declarations the AgentCore Gateway registers) and -each call dispatched to the real handler against this stack. It defaults to the Anthropic API -(`ANTHROPIC_API_KEY`); `AGENT_PROVIDER=bedrock` switches to production's transport and model id +**The agent, locally.** The container's `entrypoint.py` runs as a process on :8090 — its +`AnthropicEngine` calls `modules/agent/dev/agent.py`'s turn, with the tools DISCOVERED from their +`schema.json` (the same declarations the AgentCore Gateway registers) and each call dispatched to +the real handler against this stack, a Node tool through `node_lambda.mjs`. It runs on the Anthropic +API (`ANTHROPIC_API_KEY`); `AGENT_PROVIDER=bedrock` switches to production's transport and model id (`us.anthropic.claude-sonnet-4-6`) using your AWS creds — same messages/tool_use shape either way, so only the credentials change. What stays absent is the MCP hop: production reaches a tool through the Gateway, and locally the dispatch is in-process, so gateway-level failures (a target that will not sync, a description over 200 chars, a name collision in the shared namespace) do not appear here. +**The web chat, locally.** `modules/agent/lambdas/chat/index.mjs` runs on :8082 under +`node_lambda.mjs`, its SDK pointed at moto and at the agent process, its issuer the stand-in +(`COGNITO_ISSUER`), its Memory mirror off (`MEMORY_ID` empty — moto has no Memory data plane, so +the transcript lives in the agent process until it stops). The gerp row's `chat_url` is this door +(`LOCAL_CHAT_URL`), so the owner app's chat link works: `localhost:3000/dev/login?sub=`, open +the gerp, its chat. A turn goes browser → :8082 → `InvokeAgentRuntime` → :8090 → the reply, buffered +(the Anthropic engine does not stream), as one text chunk. `tests/agent/local/test_local_chat.py` +runs that chain against a stub agent. + ### what a request actually does `POST /webhooks/stripe` on :8080 →`_image.py` builds the payload-format-2.0 event → the real @@ -211,12 +242,13 @@ metadata as any other field. Same seam, client side: `cognito()` in the SPA posts to `CFG.COGNITO_IDP`, which on localhost is `tests/server/cognito/server.py` on :4243. It models signup, confirmation, the login-email change (`UpdateUserAttributes` → `VerifyUserAttribute`, the old address staying the login until the code -lands) and `REFRESH_TOKEN_AUTH`, accepts `000000` as every code, and mints unsigned tokens whose -`email` claim moves on verify — so `GET /api/account` writing the row from the claim runs for real. -A session that never signed up through it — the e2e helper's token, one typed into a console — is -admitted on its first call, created from the token's own `sub` and `email`; a refresh token -`rt_` names that user. The Hosted-UI sign-in itself is still real Cognito or the unsigned -JWT; the stand-in does not serve `/oauth2/*`. +lands) and `REFRESH_TOKEN_AUTH`, accepts `000000` as every code, and signs RS256 tokens (issuer +its own address, `LOCAL_COGNITO_URL`; audience `local`; the key made at start and listed at +`/.well-known/jwks.json`) whose `email` claim moves on verify — so `GET /api/account` writing the +row from the claim runs for real. `GET /token?sub=&email=` issues tokens for any sub, creating the +user on sight: `/dev/login` and the e2e login helper take theirs there. A session that never +signed up through it — a token typed into a console — is admitted on its first call the same way; +a refresh token `rt_` names that user. The stand-in does not serve `/oauth2/*`. The seller's card lambdas run in the BFF's process because `bff` depends on `per_customer` (`scripts/tags.json`) — the BFF invokes them by ARN in production, and `depends_on` is where a @@ -225,9 +257,10 @@ cross-stack call is declared. Both `image.json`s have to carry them: a stale sna ### auth -There is no authorizer here — production's APIGW validates the token before a handler sees it, and -reproducing that locally would only be reproducing AWS. Claims come from `x-debug-sub`, or from an -unverified decode of a real bearer, so a real Hosted-UI login against :3000 works. +Production's APIGW authorizer validates the token before a handler sees it. Here `_image.py` does +the same against the Cognito stand-in: a bearer's claims are read only when the stand-in signed it +(`_claims_from_bearer`, `LOCAL_COGNITO_URL` naming the issuer), and an unsigned or foreign token +carries none, so the handler answers 401. `x-debug-sub` is the header for a curl. ``` curl localhost:3000/api/gerps -H 'x-debug-sub: ' diff --git a/tests/server/_image.py b/tests/server/_image.py index 52e3c37..f7d2133 100644 --- a/tests/server/_image.py +++ b/tests/server/_image.py @@ -20,6 +20,8 @@ import pathlib import sys +sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent)) # _jwt, beside this file + from fastapi import FastAPI, Request, Response from fastapi.responses import FileResponse @@ -131,23 +133,67 @@ def make_bus(env: dict) -> str | None: return name +_JWKS: dict = {} + + +def _claims_from_bearer(token: str) -> dict: + """The bearer's claims when the Cognito stand-in at `LOCAL_COGNITO_URL` signed it, else none. A + token nothing here can check carries no claims, so the handler answers as it does behind the + gateway's authorizer, 401. The stand-in issues tokens at `/token`; `/dev/login` and the e2e + login helper take theirs there.""" + issuer = os.environ.get("LOCAL_COGNITO_URL", "").rstrip("/") + if not issuer: + return {} + import urllib.request + import _jwt + + def jwks(refresh: bool) -> dict: + if refresh or not _JWKS: + with urllib.request.urlopen(f"{issuer}/.well-known/jwks.json", timeout=2) as r: + _JWKS.clear() + _JWKS.update({k["kid"]: k for k in json.loads(r.read()).get("keys", [])}) + return _JWKS + + try: + try: + return _jwt.verify(token, jwks(False), issuer) + except ValueError: + return _jwt.verify(token, jwks(True), issuer) # the stand-in restarted with a new key + except Exception: # noqa: BLE001 — unsigned, forged, expired, malformed, or no stand-in up + return {} + + +def _node_handler(fn_name: str): + """A Node function's handler: the event goes to the local Node process (`node_lambda.mjs`, + started by scripts/local-dev.sh at `LOCAL_NODE_URL`), the answer is what the handler returned + or, for a streaming handler, what it wrote.""" + import urllib.request + base = os.environ.get("LOCAL_NODE_URL", "http://127.0.0.1:8082").rstrip("/") + + def handler(event, _context): + req = urllib.request.Request(f"{base}/invoke/{fn_name}", data=json.dumps(event).encode(), + headers={"content-type": "application/json"}, method="POST") + try: + with urllib.request.urlopen(req, timeout=120) as r: + return json.loads(r.read() or b"null") + except OSError as e: + print(f"[image] {fn_name}: the local Node process at {base} did not answer: {e}") + return {"statusCode": 502, "body": json.dumps({"error": f"the local Node process at {base} did not answer"})} + return handler + + def _event(request: Request, body: bytes, route: dict) -> dict: """The API Gateway v2 payload-format-2.0 event the deployed function receives. - Claims come from `x-debug-sub`, or from an UNVERIFIED decode of a real bearer token — the local - runner is not an authorizer, and a route the gateway protects still has to be reachable in a dev - loop. In production these arrive from the APIGW JWT authorizer and no handler ever builds them. + Claims come from a bearer the local Cognito stand-in signed (`_claims_from_bearer`), or from + `x-debug-sub`, the header for a curl. In production they arrive from the APIGW JWT authorizer and + no handler ever builds them. """ - import base64 headers = {k.lower(): v for k, v in request.headers.items()} claims = {} if auth := headers.get("authorization", ""): if auth.lower().startswith("bearer "): - try: - payload = auth.split(" ", 1)[1].split(".")[1] - claims = json.loads(base64.urlsafe_b64decode(payload + "=" * (-len(payload) % 4))) - except Exception: # noqa: BLE001 — a malformed token is simply no claims - claims = {} + claims = _claims_from_bearer(auth.split(" ", 1)[1].strip()) if sub := headers.get("x-debug-sub"): claims = {"sub": sub, "email": headers.get("x-debug-email", ""), **claims} @@ -225,6 +271,8 @@ def handler_for(fn_name): if not src: return _local_handler(fn_name) path = REPO / src / "main.py" + if not path.exists() and (REPO / src / "index.mjs").exists(): + return _node_handler(fn_name) key = (fn_name, path.stat().st_mtime_ns) if key in modules: return modules[key] diff --git a/tests/server/_jwt.py b/tests/server/_jwt.py new file mode 100644 index 0000000..9d200b8 --- /dev/null +++ b/tests/server/_jwt.py @@ -0,0 +1,59 @@ +"""RS256 tokens for the local stack. The Cognito stand-in (tests/server/cognito) signs with these; +the bff image (`_image.py`) verifies against the stand-in's jwks, and so does the chat lambda, in +Node, with the same jwks. `cryptography` only — it comes with moto.""" + +import base64 +import json +import time + +from cryptography.hazmat.primitives import hashes +from cryptography.hazmat.primitives.asymmetric import padding, rsa + + +def b64url(b: bytes) -> str: + return base64.urlsafe_b64encode(b).decode().rstrip("=") + + +def b64url_decode(s: str) -> bytes: + return base64.urlsafe_b64decode(s + "=" * (-len(s) % 4)) + + +def new_key() -> rsa.RSAPrivateKey: + return rsa.generate_private_key(public_exponent=65537, key_size=2048) + + +def jwk(key: rsa.RSAPrivateKey, kid: str) -> dict: + """The public half as a JWK, the shape `/.well-known/jwks.json` lists.""" + n = key.public_key().public_numbers() + return {"kty": "RSA", "alg": "RS256", "use": "sig", "kid": kid, + "n": b64url(n.n.to_bytes((n.n.bit_length() + 7) // 8, "big")), + "e": b64url(n.e.to_bytes((n.e.bit_length() + 7) // 8, "big"))} + + +def sign(claims: dict, key: rsa.RSAPrivateKey, kid: str) -> str: + head = b64url(json.dumps({"alg": "RS256", "typ": "JWT", "kid": kid}).encode()) + body = b64url(json.dumps(claims).encode()) + sig = key.sign(f"{head}.{body}".encode(), padding.PKCS1v15(), hashes.SHA256()) + return f"{head}.{body}.{b64url(sig)}" + + +def verify(token: str, jwks: dict, issuer: str, leeway: int = 60) -> dict: + """The claims of a token a key in `jwks` (kid → jwk) signed for `issuer`. Raises ValueError on + another alg, an unknown kid, an expired token or another issuer, and + `cryptography.exceptions.InvalidSignature` on a signature the key did not make.""" + head, body, sig = token.split(".") + header = json.loads(b64url_decode(head)) + if header.get("alg") != "RS256": + raise ValueError("unexpected alg") + k = jwks.get(header.get("kid")) + if not k: + raise ValueError("unknown signing key") + pub = rsa.RSAPublicNumbers(int.from_bytes(b64url_decode(k["e"]), "big"), + int.from_bytes(b64url_decode(k["n"]), "big")).public_key() + pub.verify(b64url_decode(sig), f"{head}.{body}".encode(), padding.PKCS1v15(), hashes.SHA256()) + claims = json.loads(b64url_decode(body)) + if claims.get("exp", 0) < time.time() - leeway: + raise ValueError("token expired") + if claims.get("iss") != issuer: + raise ValueError("bad issuer") + return claims diff --git a/tests/server/bff/dev.py b/tests/server/bff/dev.py index 7b76dac..e3c7163 100644 --- a/tests/server/bff/dev.py +++ b/tests/server/bff/dev.py @@ -5,9 +5,10 @@ there is one implementation of "an account with a card" rather than three. """ -import base64 import json import time +import os +import urllib.parse import urllib.request import uuid @@ -48,8 +49,10 @@ def _self(request: Request, sub: str, method: str, path: str, body=None): def _token(sub: str, email: str) -> str: - b64 = lambda o: base64.urlsafe_b64encode(json.dumps(o).encode()).decode().rstrip("=") # noqa: E731 - return f"{b64({'alg': 'none', 'typ': 'JWT'})}.{b64({'sub': sub, 'email': email, 'token_use': 'id', 'exp': int(time.time()) + 86400})}.local" + """An id token the Cognito stand-in signed — the only kind the bff image and the chat lambda accept.""" + base = os.environ.get("LOCAL_COGNITO_URL", "http://127.0.0.1:4243").rstrip("/") + with urllib.request.urlopen(f"{base}/token?{urllib.parse.urlencode({'sub': sub, 'email': email})}", timeout=3) as r: + return json.loads(r.read())["IdToken"] @router.get("") @@ -67,7 +70,12 @@ def index(request: Request): @verb("GET", "/login", "?sub= — a page that signs the browser in as and goes to /") def login(sub: str, email: str = ""): # every verb is sync on purpose: the self-calls need the loop free - tok = _token(sub, email or f"{sub}@localhost") + try: + tok = _token(sub, email or f"{sub}@localhost") + except OSError as e: + base = os.environ.get("LOCAL_COGNITO_URL", "http://127.0.0.1:4243") + return Response(f"the Cognito stand-in at {base} is not answering ({type(e).__name__}) — " + "it signs every local token. bash scripts/local-dev.sh --start\n", status_code=503, media_type="text/plain") return HTMLResponse(f"") diff --git a/tests/server/bff/server.py b/tests/server/bff/server.py index 646fa75..b7a6311 100644 --- a/tests/server/bff/server.py +++ b/tests/server/bff/server.py @@ -102,6 +102,9 @@ def _seed_gerps(image): # unset LOCAL_GATEWAY_URL to exercise the live gateway with a real Hosted-UI login. if os.environ.get("LOCAL_GATEWAY_URL"): row["gateway_url"] = os.environ["LOCAL_GATEWAY_URL"] + # the chat door the same way: the Node process serving modules/agent/lambdas/chat + if os.environ.get("LOCAL_CHAT_URL"): + row["chat_url"] = os.environ["LOCAL_CHAT_URL"] # The event dispatcher resolves a recipient gerp_id to its aws_account_id and then # invokes cross-account. Locally "another account" is the same process, but the row # still has to CARRY the field or the dispatcher drops the event as unroutable — which diff --git a/tests/server/cognito/server.py b/tests/server/cognito/server.py index 42ade0d..bb1b811 100644 --- a/tests/server/cognito/server.py +++ b/tests/server/cognito/server.py @@ -6,33 +6,42 @@ localhost that is this. It models signup, confirmation, the login-email change, and a token refresh, and nothing else — an action it lacks is a 400 that names itself. -Tokens are unsigned JWTs (`header.payload.local`), the shape the e2e login helper mints and the -local BFF decodes without verifying. A session that never signed up here — the e2e helper's token, -a `sessionStorage` one typed into a console — is admitted on its first attribute call: the user is -created from the token's own `sub` and `email`. Every verification code is `000000`. +Tokens are RS256 JWTs this process signs with a key it makes at start, issuer `LOCAL_COGNITO_URL` +(its own address), audience `local`. `/.well-known/jwks.json` lists the key, and that is what the +bff image (`_image.py`) and the chat lambda verify every bearer against. `GET /token?sub=&email=` +issues a token for any sub, creating the user on sight: the bff's `/dev/login` and the e2e login +helper take theirs there. A session that never signed up here — a token typed into a console — is +admitted on its first attribute call the same way: the user is created from the token's own `sub` +and `email`. Every verification code is `000000`. """ import base64 import json +import os import secrets as _secrets +import sys import time +from pathlib import Path from fastapi import FastAPI, Request from fastapi.responses import JSONResponse +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +import _jwt # noqa: E402 + app = FastAPI(title="cognito · local stand-in") # the SPA on :3000 calls this cross-origin, the way it calls the real regional endpoint, which # answers preflights itself -from fastapi.middleware.cors import CORSMiddleware +from fastapi.middleware.cors import CORSMiddleware # noqa: E402 app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]) USERS: dict = {} # sub → {email, pending_email, confirmed} BY_EMAIL: dict = {} # email → sub REFRESH: dict = {} # refresh token → sub CODE = "000000" - - -def _b64(o): - return base64.urlsafe_b64encode(json.dumps(o).encode()).decode().rstrip("=") +ISSUER = os.environ.get("LOCAL_COGNITO_URL", "http://127.0.0.1:4243").rstrip("/") +CLIENT_ID = "local" +KEY = _jwt.new_key() +KID = "local-" + _secrets.token_hex(4) def _decode(tok: str) -> dict: @@ -46,9 +55,9 @@ def _decode(tok: str) -> dict: def _tokens(sub: str) -> dict: u = USERS[sub] now = int(time.time()) - base = {"sub": sub, "email": u["email"], "iat": now, "exp": now + 3600} - return {"IdToken": f"{_b64({'alg': 'none', 'typ': 'JWT'})}.{_b64({**base, 'token_use': 'id'})}.local", - "AccessToken": f"{_b64({'alg': 'none', 'typ': 'JWT'})}.{_b64({**base, 'token_use': 'access'})}.local", + base = {"sub": sub, "email": u["email"], "iat": now, "exp": now + 3600, "iss": ISSUER} + return {"IdToken": _jwt.sign({**base, "token_use": "id", "aud": CLIENT_ID}, KEY, KID), + "AccessToken": _jwt.sign({**base, "token_use": "access", "client_id": CLIENT_ID}, KEY, KID), "TokenType": "Bearer", "ExpiresIn": 3600} @@ -56,17 +65,20 @@ def _err(kind: str, message: str, status: int = 400): return JSONResponse({"__type": kind, "message": message}, status) +def _admit(sub: str, email: str) -> str: + """A user this stand-in has not seen, created on sight.""" + if sub not in USERS: + USERS[sub] = {"email": email, "pending_email": "", "confirmed": True} + if email: + BY_EMAIL[email] = sub + return sub + + def _user_from_access(token: str): """The caller. A token this stand-in never issued still identifies one — created on sight.""" c = _decode(token or "") sub = c.get("sub") - if not sub: - return None - if sub not in USERS: - USERS[sub] = {"email": c.get("email", ""), "pending_email": "", "confirmed": True} - if c.get("email"): - BY_EMAIL[c["email"]] = sub - return sub + return _admit(sub, c.get("email", "")) if sub else None @app.post("/") @@ -145,6 +157,19 @@ async def idp(request: Request): "(tests/server/cognito/server.py)") +@app.get("/token") +def token(sub: str, email: str = ""): + """Signed tokens for `sub`, the user created on sight. The bff's `/dev/login` and the e2e login + helper sign in with these.""" + _admit(sub, email or f"{sub}@localhost") + return {**_tokens(sub), "RefreshToken": f"rt_{sub}"} + + +@app.get("/.well-known/jwks.json") +def jwks(): + return {"keys": [_jwt.jwk(KEY, KID)]} + + @app.get("/") def root(): - return {"stand_in": "cognito", "users": len(USERS), "code": CODE} + return {"stand_in": "cognito", "issuer": ISSUER, "client_id": CLIENT_ID, "kid": KID, "users": len(USERS), "code": CODE} diff --git a/tests/server/cognito/standin.py b/tests/server/cognito/standin.py new file mode 100644 index 0000000..a481e0e --- /dev/null +++ b/tests/server/cognito/standin.py @@ -0,0 +1,42 @@ +"""The Cognito stand-in as a subprocess for a test: `start()` runs `tests/server/cognito/server.py` +under uvicorn on a free port, with `LOCAL_COGNITO_URL` set to that address so the tokens it signs +name it as their issuer, and waits until it answers.""" + +import os +import socket +import subprocess +import sys +import time +import urllib.error +import urllib.request +from pathlib import Path + +HERE = Path(__file__).resolve().parent + + +def free_port() -> int: + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +def start() -> tuple[subprocess.Popen, str]: + """The process and its url. The caller terminates the process.""" + port = free_port() + url = f"http://127.0.0.1:{port}" + proc = subprocess.Popen([sys.executable, "-m", "uvicorn", "server:app", "--app-dir", str(HERE), + "--host", "127.0.0.1", "--port", str(port), "--log-level", "warning"], + env={**os.environ, "LOCAL_COGNITO_URL": url, "PYTHONPATH": str(HERE.parent)}, + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True) + for _ in range(200): + try: + urllib.request.urlopen(url + "/", timeout=1) + return proc, url + except urllib.error.HTTPError: + return proc, url + except (urllib.error.URLError, OSError): + if proc.poll() is not None: + raise RuntimeError(f"the Cognito stand-in exited: {proc.stdout.read()}") + time.sleep(0.05) + proc.terminate() + raise RuntimeError("the Cognito stand-in did not come up") diff --git a/tests/server/local/test_local_tokens.py b/tests/server/local/test_local_tokens.py new file mode 100644 index 0000000..e229ed8 --- /dev/null +++ b/tests/server/local/test_local_tokens.py @@ -0,0 +1,93 @@ +"""Every local token is one the Cognito stand-in signed. `_jwt` signs and verifies; the bff image +(`_image._claims_from_bearer`) accepts a bearer only from the stand-in at LOCAL_COGNITO_URL and +gives an unsigned, forged, expired or foreign one no claims at all.""" + +import os +import sys +import time +from pathlib import Path + +HERE = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(HERE)) +import _jwt # noqa: E402 + +ISSUER = "http://127.0.0.1:1" # a port nothing answers on: a refetch of the jwks fails fast + + +def _claims(sub="alice", **over): + now = int(time.time()) + return {"sub": sub, "email": f"{sub}@localhost", "iat": now, "exp": now + 600, "iss": ISSUER, + "token_use": "id", "aud": "local", **over} + + +def test_a_signed_token_verifies_against_its_jwk_and_nothing_else_does(): + key, other = _jwt.new_key(), _jwt.new_key() + jwks = {"k1": _jwt.jwk(key, "k1")} + tok = _jwt.sign(_claims(), key, "k1") + assert _jwt.verify(tok, jwks, ISSUER)["sub"] == "alice" + + def refused(token, why): + try: + _jwt.verify(token, jwks, ISSUER) + except Exception as e: # noqa: BLE001 + return type(e).__name__, str(e) + raise AssertionError(f"accepted a token that should be refused: {why}") + + h, p, s = tok.split(".") + none = _jwt.b64url(b'{"alg":"none","typ":"JWT"}') + assert refused(f"{none}.{p}.local", "alg none")[1] == "unexpected alg" + assert refused(_jwt.sign(_claims(), key, "k9"), "unknown kid")[1] == "unknown signing key" + assert refused(_jwt.sign(_claims(), other, "k1"), "another key, our kid")[0] == "InvalidSignature" + assert refused(_jwt.sign(_claims(exp=int(time.time()) - 120), key, "k1"), "expired")[1] == "token expired" + assert refused(_jwt.sign(_claims(iss="https://cognito-idp.us-east-1.amazonaws.com/us-east-1_x"), key, "k1"), + "another issuer")[1] == "bad issuer" + assert refused(f"{h}.{p}", "two parts")[0] == "ValueError" + + +def test_the_stand_in_issues_tokens_its_own_jwks_verifies(): + os.environ["LOCAL_COGNITO_URL"] = ISSUER + for m in [m for m in sys.modules if m.endswith("cognito.server") or m == "server"]: + del sys.modules[m] + sys.path.insert(0, str(HERE / "cognito")) + import server # noqa: E402 — tests/server/cognito/server.py, its key made at import + from fastapi.testclient import TestClient + c = TestClient(server.app) + jwks = {k["kid"]: k for k in c.get("/.well-known/jwks.json").json()["keys"]} + got = c.get("/token", params={"sub": "bob", "email": "bob@localhost"}).json() + for which, use in (("IdToken", "id"), ("AccessToken", "access")): + claims = _jwt.verify(got[which], jwks, ISSUER) + assert claims["sub"] == "bob" and claims["token_use"] == use + assert _jwt.verify(got["IdToken"], jwks, ISSUER)["aud"] == "local" + assert got["RefreshToken"] == "rt_bob" + # the user exists now: the refresh the SPA does on reload answers for it + r = c.post("/", headers={"x-amz-target": "AWSCognitoIdentityProviderService.InitiateAuth"}, + json={"AuthFlow": "REFRESH_TOKEN_AUTH", "AuthParameters": {"REFRESH_TOKEN": "rt_bob"}}) + assert _jwt.verify(r.json()["AuthenticationResult"]["IdToken"], jwks, ISSUER)["sub"] == "bob" + + +def test_the_bff_image_takes_claims_from_the_stand_ins_signature_only(): + import _image + key = _jwt.new_key() + tok = _jwt.sign(_claims(), key, "k1") + # no stand-in named: a bearer carries no claims, whoever signed it + os.environ.pop("LOCAL_COGNITO_URL", None) + _image._JWKS.clear() + _image._JWKS["k1"] = _jwt.jwk(key, "k1") + assert _image._claims_from_bearer(tok) == {} + # the stand-in named and its key known: the signed token's claims, and nothing from the others + os.environ["LOCAL_COGNITO_URL"] = ISSUER + assert _image._claims_from_bearer(tok)["sub"] == "alice" + none = _jwt.b64url(b'{"alg":"none","typ":"JWT"}') + assert _image._claims_from_bearer(f"{none}.{tok.split('.')[1]}.local") == {} + assert _image._claims_from_bearer(_jwt.sign(_claims(), _jwt.new_key(), "k1")) == {} + assert _image._claims_from_bearer("not-a-token") == {} + _image._JWKS.clear() + os.environ.pop("LOCAL_COGNITO_URL", None) + + +if __name__ == "__main__": + for _name, _fn in sorted(globals().items()): + if _name.startswith("test_") and callable(_fn): + _fn() + print(f"ok {_name}") + print("all local token tests passed") diff --git a/tests/server/local/test_node_lambda.py b/tests/server/local/test_node_lambda.py new file mode 100644 index 0000000..352a4c5 --- /dev/null +++ b/tests/server/local/test_node_lambda.py @@ -0,0 +1,167 @@ +"""`node_lambda.mjs` runs a Node lambda locally the way `_image.py` runs a Python one: the +snapshot's env applied (the shell's winning), the handler imported from source and re-imported when +the file changes, the event the deployed function receives. Two doors: the --front function as a +Function URL, streaming through; `/invoke/` answering what the handler returned or wrote.""" + +import json +import os +import socket +import subprocess +import sys +import tempfile +import time +import urllib.error +import urllib.request +from pathlib import Path + +HERE = Path(__file__).resolve().parents[1] +REPO = HERE.parents[1] +RUNNER = HERE / "node_lambda.mjs" +sys.path.insert(0, str(HERE)) + +ECHO = """ +export const handler = async (event, ctx) => ({ + statusCode: 200, headers: { "x-fn": process.env.WHO }, + body: JSON.stringify({ event, who: process.env.WHO, shell: process.env.SHELL_WINS, request: ctx.awsRequestId }), +}); +""" +STREAM = """ +async function handlerImpl(event, responseStream) { + const s = awslambda.HttpResponseStream.from(responseStream, { statusCode: 201, headers: { "content-type": "application/x-ndjson", "x-fn": process.env.WHO } }); + s.write(JSON.stringify({ type: "first" }) + "\\n"); + s.write(JSON.stringify({ path: event.rawPath, method: event.requestContext.http.method, q: event.queryStringParameters, + body: event.body, host: event.headers.host, raw: event.rawQueryString }) + "\\n"); + s.end(); +} +export const handler = (globalThis.awslambda?.streamifyResponse || ((fn) => fn))(handlerImpl); +""" + + +def _free_port(): + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +def _http(method, url, body=None, headers=None): + data = json.dumps(body).encode() if isinstance(body, (dict, list)) else (body.encode() if isinstance(body, str) else body) + req = urllib.request.Request(url, data=data, method=method, headers={"content-type": "application/json", **(headers or {})}) + try: + with urllib.request.urlopen(req, timeout=10) as r: + return r.status, dict(r.headers), r.read().decode() + except urllib.error.HTTPError as e: + return e.code, dict(e.headers), e.read().decode() + + +class _Runner: + def __init__(self, image: Path, front: str, env: dict): + self.port = _free_port() + self.proc = subprocess.Popen(["node", str(RUNNER), "--image", str(image), "--port", str(self.port), "--front", front], + env={**os.environ, **env}, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True) + self.base = f"http://127.0.0.1:{self.port}" + for _ in range(100): + try: + urllib.request.urlopen(self.base + "/invoke/none", data=b"{}", timeout=1) + except urllib.error.HTTPError: + return # 404 for an unknown function: the process answers + except (urllib.error.URLError, ConnectionError, OSError): + if self.proc.poll() is not None: + raise RuntimeError(f"node_lambda.mjs exited: {self.proc.stdout.read()}") + time.sleep(0.05) + raise RuntimeError("node_lambda.mjs did not come up") + + def stop(self): + self.proc.terminate() + self.proc.wait(timeout=5) + + +def _fixture(tmp: Path) -> Path: + (tmp / "fns/echo").mkdir(parents=True) + (tmp / "fns/stream").mkdir(parents=True) + (tmp / "fns/echo/index.mjs").write_text(ECHO) + (tmp / "fns/stream/index.mjs").write_text(STREAM) + image = {"stack": "t", "gerp": "g", "routes": [], "functions": { + "t-echo": {"src_dir": str(tmp / "fns/echo"), "env": {"WHO": "echo", "SHELL_WINS": "snapshot"}}, + "t-stream": {"src_dir": str(tmp / "fns/stream"), "env": {"WHO": "stream"}}, + "t-python": {"src_dir": "modules/accounting/lambdas/post_journal_entry", "env": {}}, + }} + (tmp / "image.json").write_text(json.dumps(image)) + return tmp / "image.json" + + +def test_invoke_runs_the_handler_under_its_env_and_the_shell_wins(): + with tempfile.TemporaryDirectory() as d: + r = _Runner(_fixture(Path(d)), "t-stream", {"SHELL_WINS": "shell"}) + try: + status, _, text = _http("POST", r.base + "/invoke/t-echo", {"x": 1}) + assert status == 200 + out = json.loads(text) + assert out["statusCode"] == 200 and out["headers"] == {"x-fn": "echo"} + body = json.loads(out["body"]) + assert body["event"] == {"x": 1} and body["who"] == "echo" and body["request"] + assert body["shell"] == "shell", "a value the shell set wins over the snapshot's" + # the src dir's last segment names the function too — a tool is known by its dir + assert json.loads(json.loads(_http("POST", r.base + "/invoke/echo", {})[2])["body"])["who"] == "echo" + status, _, text = _http("POST", r.base + "/invoke/t-nope", {}) + assert status == 404 and json.loads(text)["known"] == ["t-echo", "t-stream"], "a Python function is not a Node one" + # a streaming handler through /invoke: what it wrote, as the lambda's result + out = json.loads(_http("POST", r.base + "/invoke/t-stream", {"rawPath": "/p", "requestContext": {"http": {"method": "GET"}}, "headers": {}})[2]) + assert out["statusCode"] == 201 and out["headers"]["x-fn"] == "stream" + assert [json.loads(line) for line in out["body"].splitlines()][0] == {"type": "first"} + finally: + r.stop() + + +def test_the_front_is_the_function_url_streamed_through(): + with tempfile.TemporaryDirectory() as d: + r = _Runner(_fixture(Path(d)), "t-stream", {}) + try: + status, headers, text = _http("POST", r.base + "/api/thing?y=1&z=two", "hello") + assert status == 201 and headers["content-type"] == "application/x-ndjson" and headers["x-fn"] == "stream" + first, second = [json.loads(line) for line in text.splitlines()] + assert first == {"type": "first"} + assert second == {"path": "/api/thing", "method": "POST", "q": {"y": "1", "z": "two"}, "body": "hello", + "host": f"127.0.0.1:{r.port}", "raw": "y=1&z=two"} + finally: + r.stop() + + +def test_an_edited_handler_runs_on_the_next_request(): + with tempfile.TemporaryDirectory() as d: + tmp = Path(d) + r = _Runner(_fixture(tmp), "t-stream", {}) + try: + assert json.loads(json.loads(_http("POST", r.base + "/invoke/t-echo", {})[2])["body"])["who"] == "echo" + f = tmp / "fns/echo/index.mjs" + f.write_text(ECHO.replace("who: process.env.WHO", 'who: "edited"')) + st = f.stat() + os.utime(f, ns=(st.st_atime_ns, st.st_mtime_ns + 2_000_000)) # a later mtime whatever the clock did + assert json.loads(json.loads(_http("POST", r.base + "/invoke/t-echo", {})[2])["body"])["who"] == "edited" + finally: + r.stop() + + +def test_the_image_forwards_a_node_function_to_the_runner(): + import _image + with tempfile.TemporaryDirectory() as d: + r = _Runner(_fixture(Path(d)), "t-stream", {}) + try: + os.environ["LOCAL_NODE_URL"] = r.base + out = _image._node_handler("t-echo")({"rawPath": "/"}, None) + assert out["statusCode"] == 200 and json.loads(out["body"])["event"] == {"rawPath": "/"} + finally: + r.stop() + os.environ.pop("LOCAL_NODE_URL", None) + # with no runner up, a lambda-shaped 502 names the address, so the route answers rather than raising + os.environ["LOCAL_NODE_URL"] = "http://127.0.0.1:1" + out = _image._node_handler("t-echo")({}, None) + assert out["statusCode"] == 502 and "127.0.0.1:1" in json.loads(out["body"])["error"] + os.environ.pop("LOCAL_NODE_URL", None) + + +if __name__ == "__main__": + for _name, _fn in sorted(globals().items()): + if _name.startswith("test_") and callable(_fn): + _fn() + print(f"ok {_name}") + print("all node_lambda tests passed") diff --git a/tests/server/node_lambda.mjs b/tests/server/node_lambda.mjs new file mode 100644 index 0000000..17fa21c --- /dev/null +++ b/tests/server/node_lambda.mjs @@ -0,0 +1,166 @@ +#!/usr/bin/env node +// The Node lambdas, locally. `_image.py` runs a Python handler by importing its main.py; this runs +// an index.mjs the same way: the function's snapshotted env applied, the handler imported from +// source, called with the event the deployed function receives. scripts/local-dev.sh starts one +// per Function URL: +// +// node tests/server/node_lambda.mjs --port 8082 --front modules/agent/lambdas/chat +// +// Two doors on the port: +// POST /invoke/ +// the body is the event; the answer is what the handler returned, or for a streaming +// handler what it wrote, as {statusCode, headers, body}. dev/tools.py sends a Node tool's +// event here, and `_image.py` a Node function's +// everything else — the --front function's Function URL: the request becomes a payload-format-2.0 +// event, and what the handler writes streams back, status and headers included +// +// Env: the function's snapshotted env is applied before its import, and a value the shell set wins +// over it. That is how AWS_ENDPOINT_URL, COGNITO_ISSUER and an emptied MEMORY_ID arrive. A handler +// is re-imported when its file changes, so an edit shows on the next request. + +import { randomUUID } from "node:crypto"; +import { existsSync, readFileSync, statSync } from "node:fs"; +import http from "node:http"; +import path from "node:path"; +import { fileURLToPath, pathToFileURL } from "node:url"; + +const REPO = path.resolve(path.dirname(fileURLToPath(import.meta.url)), "..", ".."); +const argv = process.argv.slice(2); +const arg = (name, dflt) => { const i = argv.indexOf(`--${name}`); return i >= 0 ? argv[i + 1] : dflt; }; +const PORT = Number(arg("port", 8082)); +const IMAGE = path.resolve(REPO, arg("image", "tests/server/per_customer/image.json")); +const image = JSON.parse(readFileSync(IMAGE, "utf8")); + +// the Node functions in the image: a src dir holding an index.mjs +const functions = Object.entries(image.functions) + .filter(([, cfg]) => cfg.src_dir && existsSync(path.resolve(REPO, cfg.src_dir, "index.mjs"))) + .map(([name, cfg]) => ({ name, src: cfg.src_dir, file: path.resolve(REPO, cfg.src_dir, "index.mjs"), env: cfg.env || {} })); +const byKey = new Map(); +for (const f of functions) { byKey.set(f.name, f); byKey.set(f.src, f); byKey.set(path.basename(f.src), f); } +const FRONT = arg("front", ""); +const front = FRONT ? byKey.get(FRONT) : null; +if (FRONT && !front) { console.error(`[node] no Node function ${FRONT} in ${IMAGE}`); process.exit(1); } + +// the shell's env wins over the snapshot's +const shell = new Set(Object.keys(process.env)); +function applyEnv(fn) { + for (const [k, v] of Object.entries(fn.env)) if (!shell.has(k)) process.env[k] = v; +} + +// what the streaming-enabled Lambda runtime provides: `streamifyResponse` wraps an +// (event, responseStream) handler, `HttpResponseStream.from` puts the status and headers on the stream +globalThis.awslambda = { + streamifyResponse: (fn) => fn, + HttpResponseStream: { from(stream, meta) { stream.open(meta?.statusCode || 200, meta?.headers || {}); return stream; } }, +}; + +const loaded = new Map(); // name → {mtime, handler} +async function handlerOf(fn) { + const mtime = statSync(fn.file).mtimeMs; + const have = loaded.get(fn.name); + if (have && have.mtime === mtime) return have.handler; + applyEnv(fn); + const mod = await import(`${pathToFileURL(fn.file).href}?v=${mtime}`); + loaded.set(fn.name, { mtime, handler: mod.handler }); + return mod.handler; +} + +// The handler's second argument. A streaming handler writes to it; a plain one reads it as the +// context. One object serves both, so neither handler shape has to be told apart. +function context(fn) { + return { awsRequestId: randomUUID(), functionName: fn.name, getRemainingTimeInMillis: () => 60_000 }; +} +function streamTo(res, fn) { // the front: written through to the browser + return { + ...context(fn), started: false, + open(status, headers) { this.started = true; res.writeHead(status, headers); }, + write(chunk) { res.write(chunk); }, + end(chunk) { res.end(chunk); }, + }; +} +function collector(fn) { // /invoke: collected, then answered as the lambda's result + const chunks = []; + return { + ...context(fn), started: false, status: 200, headers: {}, + open(status, headers) { this.started = true; this.status = status; this.headers = headers; }, + write(chunk) { chunks.push(Buffer.from(chunk)); }, + end(chunk) { if (chunk) chunks.push(Buffer.from(chunk)); }, + result() { return { statusCode: this.status, headers: this.headers, body: Buffer.concat(chunks).toString("utf8") }; }, + }; +} + +const readBody = (req) => new Promise((resolve) => { + const parts = []; + req.on("data", (c) => parts.push(c)); + req.on("end", () => resolve(Buffer.concat(parts))); +}); + +// the payload-format-2.0 event a Function URL hands its function +function urlEvent(req, body) { + const url = new URL(req.url, `http://${req.headers.host || "localhost"}`); + const headers = { ...req.headers }; + const event = { + version: "2.0", routeKey: "$default", rawPath: url.pathname, rawQueryString: url.search.slice(1), headers, + requestContext: { + http: { method: req.method, path: url.pathname, protocol: "HTTP/1.1", + sourceIp: req.socket.remoteAddress || "", userAgent: headers["user-agent"] || "" }, + domainName: headers.host || "", requestId: randomUUID(), timeEpoch: Date.now(), + }, + isBase64Encoded: false, + }; + if (url.search.length > 1) event.queryStringParameters = Object.fromEntries(url.searchParams); + if (headers.cookie) event.cookies = headers.cookie.split(/;\s*/); + if (body.length) event.body = body.toString("utf8"); + return event; +} + +function sendJson(res, status, obj) { + res.writeHead(status, { "content-type": "application/json" }); + res.end(JSON.stringify(obj)); +} + +// a lambda-shaped result → the response the Function URL would send +function sendResult(res, r) { + if (r === undefined || r === null) return sendJson(res, 200, null); + if (typeof r !== "object" || !("statusCode" in r)) return sendJson(res, 200, r); + const headers = { ...(r.headers || {}) }; + if (r.cookies) headers["set-cookie"] = r.cookies; + res.writeHead(r.statusCode || 200, headers); + return res.end(r.isBase64Encoded ? Buffer.from(r.body || "", "base64") : (r.body ?? "")); +} + +const server = http.createServer(async (req, res) => { + const body = await readBody(req); + const t0 = Date.now(); + // a constant format string: the url is a request's own text and is printed as a value + const log = (fn, status) => console.log("[node] %s %s %s %d %dms", fn ? fn.name : "-", req.method, req.url, status, Date.now() - t0); + let fn = null; + try { + const m = req.url.match(/^\/invoke\/([^/?]+)/); + if (m) { + fn = byKey.get(decodeURIComponent(m[1])); + if (!fn) { sendJson(res, 404, { error: `no Node function ${m[1]}`, known: functions.map((f) => f.name) }); return log(null, 404); } + const event = body.length ? JSON.parse(body.toString("utf8")) : {}; + const out = collector(fn); + const r = await (await handlerOf(fn))(event, out); + sendJson(res, 200, out.started ? out.result() : (r === undefined ? null : r)); + return log(fn, 200); + } + if (!front) { sendJson(res, 404, { error: "no --front function", invoke: functions.map((f) => `/invoke/${f.name}`) }); return log(null, 404); } + fn = front; + const out = streamTo(res, fn); + const r = await (await handlerOf(fn))(urlEvent(req, body), out); + if (!out.started) sendResult(res, r); + return log(fn, res.statusCode); + } catch (e) { + console.error("[node] %s %s %s failed:", fn ? fn.name : "-", req.method, req.url, e); + if (!res.headersSent) sendJson(res, 500, { error: `${e?.name || "Error"}: ${e?.message || e}` }); + else res.end(); + return log(fn, 500); + } +}); + +server.listen(PORT, "127.0.0.1", () => { + console.log(`[node] :${PORT} · ${functions.length} Node functions: ${functions.map((f) => f.name).join(", ")}` + + (front ? ` · front ${front.name} (${front.src})` : "")); +});