diff --git a/packages/agentenv-protocol/src/agentenv_protocol/website/__init__.py b/packages/agentenv-protocol/src/agentenv_protocol/website/__init__.py new file mode 100644 index 00000000..5783c22f --- /dev/null +++ b/packages/agentenv-protocol/src/agentenv_protocol/website/__init__.py @@ -0,0 +1,30 @@ +"""The website SDK: a site's server, meeting ``urn:agentenv:website/v1``. Subclass WebsiteEnvironment, declare the +site with ``@website(domains=[...], static="dist/")``, mark its API with ``@api.get/post/put/patch/delete(path)``, and +``serve()`` it; it needs only the plain package, and uvicorn to serve. The class is optional: a server in any language +that meets the contract works, which ``python -m agentenv_protocol.website.conformance `` checks.""" + +from __future__ import annotations + +from .clock import CLOCK_EXTENSION_URI, UNIVERSE_NOW_HEADER, UniverseClock +from .environment import ( + DEFAULT_HEALTH_PATH, + SESSION_PATH, + STATE_PATH, + WEBSITE_EXTENSION_URI, + WebsiteDeclaration, + WebsiteEnvironment, + ready_check, + website, +) +from .errors import ApiError, BadRequest, Conflict, Forbidden, NotFound, Unauthorized +from .routes import ROLE_KEY_ENV, ROLE_KEY_HEADER, ApiContext, ApiRoute, api, reserved_path +from .session import SESSION_COOKIE, SessionSigner +from .store import Collections + +__all__ = [ + "WebsiteEnvironment", "WebsiteDeclaration", "website", "api", "ApiContext", "ApiRoute", "ready_check", + "ApiError", "BadRequest", "Unauthorized", "Forbidden", "NotFound", "Conflict", + "Collections", "UniverseClock", "SessionSigner", "reserved_path", + "WEBSITE_EXTENSION_URI", "CLOCK_EXTENSION_URI", "UNIVERSE_NOW_HEADER", "DEFAULT_HEALTH_PATH", "SESSION_PATH", + "STATE_PATH", "SESSION_COOKIE", "ROLE_KEY_ENV", "ROLE_KEY_HEADER", +] diff --git a/packages/agentenv-protocol/src/agentenv_protocol/website/clock.py b/packages/agentenv-protocol/src/agentenv_protocol/website/clock.py new file mode 100644 index 00000000..598699cf --- /dev/null +++ b/packages/agentenv-protocol/src/agentenv_protocol/website/clock.py @@ -0,0 +1,100 @@ +"""The universe's clock as a site reads it: ``clock/v1``'s ``sync_time`` hands the server the gateway's ``/clock/time`` +URL, as it hands it to MCP servers, and the site's time then follows that clock's ``{"virtual_time": }``, +read at most once a second and extrapolated between reads. The rate it extrapolates at is one two successive intervals +between reads agreed on, so a clock set or advanced between two reads is taken for a jump, not a speed; until two agree +it is real time's. An unsynced site reads wall time. Every answer of a site carries the time it answered at as ``X-Universe-Now``, +RFC3339 in UTC.""" +from __future__ import annotations + +import asyncio +import datetime +import json +import logging +import time + +import httpx + +logger = logging.getLogger(__name__) + +CLOCK_EXTENSION_URI = "urn:agentenv:clock/v1" +UNIVERSE_NOW_HEADER = "X-Universe-Now" +REFRESH_S = 1.0 +READ_TIMEOUT_S = 5.0 + + +def iso(moment: datetime.datetime) -> str: + return moment.astimezone(datetime.timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z") + + +def parse_time(value: str) -> datetime.datetime: + moment = datetime.datetime.fromisoformat(value.replace("Z", "+00:00")) + return moment if moment.tzinfo else moment.replace(tzinfo=datetime.timezone.utc) + + +class UniverseClock: + """The time a site answers with: wall time until ``sync`` names a gateway clock, that clock's after.""" + + def __init__(self) -> None: + self.url: str | None = None + self._virtual: datetime.datetime | None = None + self._read_at = 0.0 + self._tried_at = 0.0 + self._rate = 1.0 + self._measured: float | None = None + self._lock: asyncio.Lock | None = None + + @property + def synced(self) -> bool: + return self._virtual is not None + + def now(self) -> datetime.datetime: + if self._virtual is None: + return datetime.datetime.now(datetime.timezone.utc) + return self._virtual + datetime.timedelta(seconds=(time.monotonic() - self._read_at) * self._rate) + + async def sync(self, url: str) -> datetime.datetime: + """Follow the clock at ``url`` from now on; raises when it can't be read, keeping the clock it followed.""" + virtual = await _read(url) + self.url, self._virtual, self._rate, self._measured = url, virtual, 1.0, None + self._read_at = self._tried_at = time.monotonic() + return virtual + + async def refresh(self) -> None: + """Read the synced clock again when the last read is a second old; a failed read keeps extrapolating.""" + if self.url is None or time.monotonic() - self._tried_at < REFRESH_S: + return + self._lock = self._lock or asyncio.Lock() + async with self._lock: + started = time.monotonic() + if started - self._tried_at < REFRESH_S: + return + self._tried_at = started + try: + virtual = await _read(self.url) + except Exception as e: + logger.debug("reading the universe's clock at %s failed: %r", self.url, e) + return + read_at = time.monotonic() + if self._virtual is not None and read_at > self._read_at: + measured = max(0.0, (virtual - self._virtual).total_seconds() / (read_at - self._read_at)) + if self._measured is not None and abs(measured - self._measured) <= 0.1 * max(measured, self._measured, 1.0): + self._rate = measured + self._measured = measured + self._virtual, self._read_at = virtual, read_at + + +async def _read(url: str) -> datetime.datetime: + async with httpx.AsyncClient() as client: + response = await client.get(url, timeout=READ_TIMEOUT_S) + response.raise_for_status() + return parse_time(response.json()["virtual_time"]) + + +def clock_patch_script(now: datetime.datetime) -> bytes: + """An inline script that shifts the page's ``Date`` to the universe's time ``now``, and follows the + ``X-Universe-Now`` header of every ``fetch`` the page makes after.""" + return ("" % (int(now.timestamp() * 1000), json.dumps(UNIVERSE_NOW_HEADER))).encode() diff --git a/packages/agentenv-protocol/src/agentenv_protocol/website/conformance.py b/packages/agentenv-protocol/src/agentenv_protocol/website/conformance.py new file mode 100644 index 00000000..f6f3e4ea --- /dev/null +++ b/packages/agentenv-protocol/src/agentenv_protocol/website/conformance.py @@ -0,0 +1,263 @@ +"""Check a running site's server against ``urn:agentenv:website/v1``, whatever it's written in: +``python -m agentenv_protocol.website.conformance `` prints each check and exits 1 when one failed, 2 when the +server couldn't be reached. Everything goes over HTTP: the card, ``/`` under a made-up host, the declared health path, +``X-Universe-Now``, a data-plane round trip (data/get, data/reset, data/add of what data/get returned, data/get again, +which leaves the server's state as it found it), ``clock/v1``'s ``sync_time`` against a clock the checker serves, and +the forwarded headers wherever an answer shows them, on ``/`` and on a GET route with its trailing slash toggled, which +a server may redirect. The clock check leaves the server following a clock that is gone +once the check ends, so a server it ran against keeps extrapolating from that clock's time until it's synced again. The checker's clock starts at ``CLOCK_START``, far enough from wall time that a synced +server shows it, and a synced server's ``X-Universe-Now`` may be ``CLOCK_SLACK_S`` seconds from it.""" +from __future__ import annotations + +import argparse +import contextlib +import dataclasses +import datetime +import json +import sys +import threading +import time +import uuid +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Callable, Iterator, Literal + +import httpx + +from ..client import find_extension, find_extension_method +from ..types import METHOD_ADD, METHOD_GET, METHOD_RESET, RPC_PATH, WELL_KNOWN_PATH +from .clock import CLOCK_EXTENSION_URI, UNIVERSE_NOW_HEADER, iso, parse_time +from .environment import DEFAULT_HEALTH_PATH, WEBSITE_EXTENSION_URI + +Status = Literal["passed", "failed", "skipped"] +CLOCK_START = datetime.datetime(2031, 5, 6, 7, 8, 9, tzinfo=datetime.timezone.utc) +CLOCK_SLACK_S = 120 + + +@dataclasses.dataclass(frozen=True) +class Result: + name: str + status: Status + detail: str = "" + + @property + def passed(self) -> bool: + return self.status != "failed" + + +def check(base_url: str, *, client: httpx.Client | None = None, clock_host: str = "127.0.0.1", + timeout: float = 30) -> list[Result]: + """Each check's result against the server at ``base_url``. ``clock_host`` is the address the server reaches the + checker's clock at. Raises httpx's error when the server can't be reached at all.""" + own = client is None + client = client or httpx.Client(timeout=timeout) + try: + return _Checker(base_url.rstrip("/"), client, clock_host).run() + finally: + if own: + client.close() + + +class _Checker: + def __init__(self, base_url: str, client: httpx.Client, clock_host: str) -> None: + self.base_url, self.client, self.clock_host = base_url, client, clock_host + self.responses: list[httpx.Response] = [] + + def get(self, path: str, **kwargs) -> httpx.Response: + return self._seen(self.client.get(self.base_url + path, **kwargs)) + + def _seen(self, response: httpx.Response) -> httpx.Response: + self.responses.append(response) + return response + + def run(self) -> list[Result]: + response = self.get(WELL_KNOWN_PATH) + try: + card = response.json() if response.status_code == 200 else {} + except ValueError: + card = {} + site = (find_extension(card, WEBSITE_EXTENSION_URI) or {}).get("params") or {} + results = [self._card(response, site)] + health = site.get("health") if isinstance(site.get("health"), str) else DEFAULT_HEALTH_PATH + for name, step in (("any_host", self._any_host), ("health", lambda: self._health(health)), + ("universe_now", self._universe_now), ("data_plane", lambda: self._data_plane(card)), + ("clock", lambda: self._clock(card, health)), ("forwarded", lambda: self._forwarded(site)), + ("cookies", lambda: self._cookies(site))): + try: + status, detail = step() + except Exception as e: + status, detail = "failed", f"{type(e).__name__}: {e}" + results.append(Result(name, status, detail)) + return results + + def _card(self, response: httpx.Response, site: dict) -> Result: + if response.status_code != 200: + return Result("card", "failed", f"{WELL_KNOWN_PATH} answered {response.status_code}") + problems = [] + if not site: + problems.append(f"declares no {WEBSITE_EXTENSION_URI} with params") + if not isinstance(site.get("domains"), list): + problems.append("its params have no domains list") + if not (isinstance(site.get("health"), str) and site["health"].startswith("/")): + problems.append("its params have no health path") + routes = site.get("routes") + if not isinstance(routes, list) or not all(isinstance(r, dict) and {"method", "path", "name"} <= set(r) + for r in routes): + problems.append("its params' routes aren't a list of {method, path, name}") + return Result("card", "failed" if problems else "passed", "; ".join(problems)) + + def _any_host(self) -> tuple[Status, str]: + host = f"conformance-{uuid.uuid4().hex[:8]}.invalid" + response = self.get("/", headers={"Host": host}) + kind = response.headers.get("content-type", "") + if response.status_code == 200 and kind.startswith("text/html"): + return "passed", f"/ served HTML for Host {host}" + return "failed", f"/ for Host {host} answered {response.status_code} {kind or 'without a content type'}" + + def _health(self, health: str) -> tuple[Status, str]: + response = self.get(health) + return ("passed" if response.status_code == 200 else "failed"), f"{health} answered {response.status_code}" + + def _universe_now(self) -> tuple[Status, str]: + missing = [str(r.url.path) for r in self.responses if UNIVERSE_NOW_HEADER not in r.headers] + if missing: + return "failed", f"no {UNIVERSE_NOW_HEADER} on {', '.join(missing)}" + for response in self.responses: + parse_time(response.headers[UNIVERSE_NOW_HEADER]) + return "passed", f"on all {len(self.responses)} answers" + + def _rpc(self, method: str, params: dict | None = None) -> dict: + response = self._seen(self.client.post(self.base_url + RPC_PATH, json={"jsonrpc": "2.0", "id": method, + "method": method, "params": params or {}})) + body = response.json() + if "error" in body: + raise RuntimeError(f"{method} answered {body['error']}") + return body.get("result") or {} + + def _data_plane(self, card: dict) -> tuple[Status, str]: + operations = (card.get("capabilities") or {}).get("operations") + missing = [m for m in (METHOD_GET, METHOD_RESET, METHOD_ADD) if operations is not None and m not in operations] + if missing: + return "failed", f"the card's operations lack {', '.join(missing)}" + exported = self._rpc(METHOD_GET).get("parts") or [] + self._rpc(METHOD_RESET) + self._rpc(METHOD_GET) + if not exported: + return "passed", "data/get, data/reset and data/get answered; data/get returned nothing to add back" + self._rpc(METHOD_ADD, {"parts": exported}) + again = self._rpc(METHOD_GET).get("parts") or [] + if again != exported: + return "failed", "data/get after data/reset and data/add of its own export returned other data" + return "passed", f"{len(exported)} part(s) round-tripped through reset and add" + + def _clock(self, card: dict, health: str) -> tuple[Status, str]: + if find_extension(card, CLOCK_EXTENSION_URI) is None: + return "failed", f"the card declares no {CLOCK_EXTENSION_URI}" + params = find_extension(card, CLOCK_EXTENSION_URI).get("params") or {} + method = find_extension_method(card, CLOCK_EXTENSION_URI, next(iter(params.get("methods") or {}), "sync_time")) + if method is None or not method.get("endpoint"): + return "failed", f"{CLOCK_EXTENSION_URI} advertises no sync_time endpoint" + with _clock_server(self.clock_host) as (url, virtual_now): + response = self._seen(self.client.request(method["method"], self.base_url + method["endpoint"], + json={"env_get_time_url": url})) + if response.status_code >= 400: + return "failed", f"sync_time answered {response.status_code}: {response.text[:200]}" + seen = self.get(health).headers.get(UNIVERSE_NOW_HEADER) + expected = virtual_now() + if seen is None: + return "failed", f"no {UNIVERSE_NOW_HEADER} after sync_time" + off = abs((parse_time(seen) - expected).total_seconds()) + if off > CLOCK_SLACK_S: + return "failed", f"after sync_time {UNIVERSE_NOW_HEADER} is {seen}, {off:.0f}s from the clock's {iso(expected)}" + return "passed", f"{UNIVERSE_NOW_HEADER} followed the clock to {seen}" + + def _forwarded(self, site: dict) -> tuple[Status, str]: + internal = f"internal-{uuid.uuid4().hex[:8]}.invalid" + forwarded = f"site-{uuid.uuid4().hex[:8]}.example" + headers = {"Host": internal, "X-Forwarded-Host": forwarded, "X-Forwarded-Proto": "https"} + paths = ["/", *_slash_toggled(site.get("routes"))] + for path in paths: + response = self.get(path, headers=headers, follow_redirects=False) + location = response.headers.get("location", "") + shown = [where for where, text in (("body", response.text), ("Location", location)) if internal in text] + if shown: + return "failed", f"the {' and '.join(shown)} of {path} named the internal Host {internal}, not {forwarded}" + if location.startswith(f"http://{forwarded}"): + return "failed", f"the Location of {path} is {location}, not https as X-Forwarded-Proto said" + return "passed", f"no answer to {', '.join(paths)} named the internal Host" + + def _cookies(self, site: dict) -> tuple[Status, str]: + if isinstance(site.get("session_path"), str): + self.get(site["session_path"]) + cookies = [c for r in self.responses for c in r.headers.get_list("set-cookie")] + if not cookies: + return "skipped", "no answer set a cookie" + scoped = [c.split("=", 1)[0] for c in cookies if any(a.strip().lower().startswith("domain=") for a in c.split(";"))] + if scoped: + return "failed", f"cookies tied to a domain: {', '.join(sorted(set(scoped)))}" + return "passed", f"{len(cookies)} cookie(s), none tied to a domain" + + +def _slash_toggled(routes: object) -> list[str]: + """The first GET route without path parameters, with its trailing slash added or taken off: a path a server + may redirect to the route's own, which shows the address the redirect names.""" + for route in routes if isinstance(routes, list) else []: + path = route.get("path") if isinstance(route, dict) and route.get("method") == "GET" else None + if isinstance(path, str) and path.startswith("/") and path != "/" and "{" not in path: + return [path[:-1] if path.endswith("/") else path + "/"] + return [] + + +@contextlib.contextmanager +def _clock_server(host: str) -> Iterator[tuple[str, Callable[[], datetime.datetime]]]: + """A gateway-shaped ``/clock/time`` running from ``CLOCK_START`` at real speed, and its time now.""" + started = time.monotonic() + + def virtual_now() -> datetime.datetime: + return CLOCK_START + datetime.timedelta(seconds=time.monotonic() - started) + + class Clock(BaseHTTPRequestHandler): + def do_GET(self) -> None: + body = json.dumps({"virtual_time": iso(virtual_now())}).encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, *args) -> None: + return None + + server = ThreadingHTTPServer(("127.0.0.1" if host in ("127.0.0.1", "localhost") else "0.0.0.0", 0), Clock) + threading.Thread(target=server.serve_forever, daemon=True).start() + try: + address = f"[{host}]" if ":" in host else host + yield f"http://{address}:{server.server_port}/clock/time", virtual_now + finally: + server.shutdown() + server.server_close() + + +_MARK = {"passed": "PASS", "failed": "FAIL", "skipped": "----"} + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(prog="python -m agentenv_protocol.website.conformance", + description="Check a running site's server against urn:agentenv:website/v1.") + parser.add_argument("url", help="where the server answers, such as http://localhost:8000") + parser.add_argument("--clock-host", default="127.0.0.1", help="the address the server reaches this checker at") + parser.add_argument("--timeout", type=float, default=30, help="seconds for each request") + args = parser.parse_args(argv) + try: + results = check(args.url, clock_host=args.clock_host, timeout=args.timeout) + except httpx.TransportError as e: + print(f"couldn't reach {args.url}: {type(e).__name__}: {e}", file=sys.stderr) + return 2 + for result in results: + print(f"{_MARK[result.status]} {result.name}{': ' + result.detail if result.detail else ''}") + failed = sum(not r.passed for r in results) + print(f"{args.url}: {len(results) - failed} passed or skipped, {failed} failed") + return 1 if failed else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/packages/agentenv-protocol/src/agentenv_protocol/website/environment.py b/packages/agentenv-protocol/src/agentenv_protocol/website/environment.py new file mode 100644 index 00000000..4fba0c48 --- /dev/null +++ b/packages/agentenv-protocol/src/agentenv_protocol/website/environment.py @@ -0,0 +1,392 @@ +"""WebsiteEnvironment: an AgentEnvEnvironment that is a site's server, meeting ``urn:agentenv:website/v1``. It serves +plain HTTP on ``PORT`` for any host name: its ``@api`` routes, its static folder at ``/`` with ``index.html`` for the +pages a single-page app routes itself, its health path, and the data plane and card the base class serves. Every answer +carries ``X-Universe-Now``, the universe's time, which ``self.now()`` reads and ``clock/v1``'s ``sync_time`` ties to +the gateway's virtual clock. Without data hooks of its own a site keeps its state in ``self.store``. + +Two routes are built in, each replaced by a site's own ``@api`` route at its path: ``/api/session``, which a frontend +calls to learn who is signed in, signing the acting user in when nobody is (GET), or to sign out (DELETE); and, for a +store-backed site, ``/go``, its ``{initial_state, current_state, state_diff}`` as the website state verifier reads it.""" +from __future__ import annotations + +import inspect +import json +import os +import re +from pathlib import Path +from typing import Any, ClassVar, NamedTuple, Optional + +from starlette.applications import Starlette +from starlette.datastructures import MutableHeaders +from starlette.middleware import Middleware +from starlette.requests import Request +from starlette.responses import FileResponse, JSONResponse, Response +from starlette.routing import Route + +from ..agent_env_environment import ( + _OP_ATTR, + AgentEnvEnvironment, + AgentEnvStarletteApplication, + _marked_members, + _maybe_await, + add_data, + extension, + get_data, + reset_data, +) +from ..types import DataPart +from .clock import CLOCK_EXTENSION_URI, UNIVERSE_NOW_HEADER, UniverseClock, clock_patch_script, iso +from .errors import NotFound +from .routes import ApiContext, ApiRoute, endpoint, error_response, openapi, openapi_path, request_role, reserved_path, routes_of +from .session import SESSION_COOKIE, SessionSigner, expires_at +from .store import Collections + +WEBSITE_EXTENSION_URI = "urn:agentenv:website/v1" +DEFAULT_HEALTH_PATH = "/api/health" +SESSION_PATH = "/api/session" +STATE_PATH = "/go" +_WEBSITE_ATTR = "_aee_website" +_READY_ATTR = "_aee_website_ready" +_DEFAULT_HOOK_ATTR = "_aee_default_store_hook" +_DEFAULT_HOOKS = {"reset_data": "_store_reset", "add_data": "_store_add", "get_data": "_store_get"} +_HEAD = re.compile(rb"]*>", re.IGNORECASE) + + +class WebsiteDeclaration(NamedTuple): + """What ``@website`` declares: the site's domains, its static folder and seed data file (relative to the class's + file), its health path, and whether its HTML pages read the universe's time.""" + + domains: tuple[str, ...] = () + static: Optional[str] = None + health: str = DEFAULT_HEALTH_PATH + clock_patch: bool = False + data: Optional[str] = None + + +def website(_cls: type | None = None, *, domains: list[str] | tuple[str, ...] = (), static: str | None = None, + health: str = DEFAULT_HEALTH_PATH, clock_patch: bool = False, data: str | None = None) -> Any: + """Declare a WebsiteEnvironment's site. ``data`` names a JSON file the default store starts with.""" + if _cls is not None: + raise TypeError("use @website(...) with parentheses") + if not health.startswith("/") or health == "/" or reserved_path(health): + raise ValueError(f"health is a path of its own under '/', not {health!r}") + + def deco(cls: type) -> type: + if not (isinstance(cls, type) and issubclass(cls, WebsiteEnvironment)): + raise TypeError(f"@website declares a WebsiteEnvironment subclass, not {cls!r}") + setattr(cls, _WEBSITE_ATTR, WebsiteDeclaration(tuple(domains), static, health, clock_patch, data)) + _check_health(cls) + return cls + + return deco + + +def ready_check(fn: Any) -> Any: + """Mark a method that says whether the site is usable yet: the health path answers 200 once every one returns a + truthy value, and 503 while one returns a falsy one or raises.""" + setattr(fn, _READY_ATTR, True) + return fn + + +class WebsiteEnvironment(AgentEnvEnvironment): + """Subclass it, declare the site with ``@website(...)``, mark its API with ``@api``, and ``serve()`` it. A + subclass's ``__init__`` calls ``super().__init__()``. ``app`` may be a Starlette or FastAPI app answering what + ``@api`` and the static folder don't (streaming, uploads); its lifespan is not run.""" + + app: ClassVar[Any] = None + asgi: Any = None + + def __init_subclass__(cls, **kwargs: Any) -> None: + super().__init_subclass__(**kwargs) + routes_of(cls) + own = {getattr(member, _OP_ATTR) for klass in cls.__mro__ for member in vars(klass).values() + if getattr(member, _OP_ATTR, None) is not None and not getattr(member, _DEFAULT_HOOK_ATTR, False)} + for op, name in _DEFAULT_HOOKS.items(): + if op in own: + setattr(cls, name, None) + + def __init__(self) -> None: + declaration = self.declaration() + self.store = Collections() + if declaration.data is not None: + self.store.load(json.loads(self._site_path(declaration.data).read_text(encoding="utf-8"))) + self.store.mark_initial() + self.clock = UniverseClock() + self._signer = SessionSigner() + + @classmethod + def declaration(cls) -> WebsiteDeclaration: + return getattr(cls, _WEBSITE_ATTR, None) or WebsiteDeclaration() + + @classmethod + def routes(cls) -> list[ApiRoute]: + return routes_of(cls) + + @property + def database_url(self) -> str | None: + """Where the site keeps its task state: its own schema of the env's database.""" + return os.environ.get("DATABASE_URL") + + @property + def uses_store(self) -> bool: + """Whether the site's data plane exports ``self.store``: it declares no ``@get_data`` of its own.""" + return type(self)._store_get is not None + + def now(self) -> Any: + """The universe's time, an aware UTC datetime: the gateway's virtual clock once synced, else wall time.""" + return self._clock().now() + + def acting_user(self) -> str | None: + """Who a browser with no session is signed in as: the store's ``current_user`` unless a site says else.""" + return self.store.current_user if hasattr(self, "store") else None + + def context(self, request: Request) -> ApiContext: + """The ``ApiContext`` of ``request``, for a mounted ``app``'s routes as for ``@api`` ones.""" + session = self._session_signer().decode(request.cookies.get(SESSION_COOKIE)) + user = session["user"] if session is not None else self.acting_user() + return ApiContext(request, user, request_role(request), session, self._session_signer()) + + def create_app(self) -> Starlette: + if self._mounted: + raise RuntimeError(f"{type(self).__name__} is already mounted") + declaration = self.declaration() + _check_health(type(self)) + api_routes = self.routes() + taken = {openapi_path(r.path) for r in api_routes} + root = self._site_path(declaration.static) if declaration.static is not None else None + asgi = Starlette(middleware=[Middleware(_Forwarded), Middleware(_Universe, env=self)]) + AgentEnvStarletteApplication(self._build_card(), self).add_routes_to_app(asgi) + asgi.router.routes.append(Route(declaration.health, self._health, methods=["GET"])) + asgi.router.routes.extend(Route(r.path, endpoint(getattr(self, r.fn.__name__), r, self.context), + methods=[r.method]) for r in api_routes) + if SESSION_PATH not in taken: + asgi.router.routes.append(Route(SESSION_PATH, self._session, methods=["GET", "DELETE"])) + if self.uses_store and STATE_PATH not in taken: + asgi.router.routes.append(Route(STATE_PATH, self._state, methods=["GET"])) + asgi.router.default = _Fallback(root, type(self).app) + self.asgi = asgi + self._mounted = True + return asgi + + def mount(self, app: Any) -> Any: + raise TypeError(f"{type(self).__name__} builds its own app with create_app(); set app = ... to serve more") + + def serve(self) -> None: + """Serve the site on ``PORT`` (8000 by default) with uvicorn, of the ``agent`` or ``app`` extra.""" + try: + import uvicorn + except ImportError as e: + raise ImportError("WebsiteEnvironment.serve needs uvicorn: pip install 'uvicorn>=0.30'") from e + uvicorn.run(self.asgi or self.create_app(), host=os.environ.get("HOST", "0.0.0.0"), + port=int(os.environ.get("PORT", "8000"))) + + def extension_params(self, uri: str) -> dict: + if uri == WEBSITE_EXTENSION_URI: + declaration = self.declaration() + return {"domains": list(declaration.domains), "health": declaration.health, + "routes": [r.advertised() for r in self.routes()], "session_cookie": SESSION_COOKIE, + "session_path": SESSION_PATH} + return super().extension_params(uri) + + @extension(CLOCK_EXTENSION_URI, description="Sync this site to the gateway's virtual clock.") + async def sync_time(self, env_get_time_url: str) -> dict: + """Follow the gateway's virtual time at ``env_get_time_url`` from now on.""" + virtual = await self._clock().sync(env_get_time_url) + return {"env_get_time_url": env_get_time_url, "gateway_virtual_time": iso(virtual)} + + @extension(WEBSITE_EXTENSION_URI, op="openapi", method="GET", description=( + "A site's server: domains, its health path and its @api routes by name; openapi returns the OpenAPI document " + "of its routes, and session the cookies a browser adds to start signed in as user, or signed out without one.")) + async def _website_openapi(self) -> dict: + name = self._application.environment_card.name if self._application is not None else type(self).__name__ + return openapi(name, self.routes()) + + @extension(WEBSITE_EXTENSION_URI, op="session") + async def _website_session(self, user: Optional[str] = None) -> dict: + cookie = self._session_signer().cookie(user) + return {"user": user, "cookies": [cookie], "expires_at": expires_at(cookie["expires"])} + + @reset_data + async def _store_reset(self) -> None: + self.store.reset() + + @add_data + async def _store_add(self, parts: list) -> None: + self.store.load_parts(parts) + self.store.mark_initial() + + @get_data + async def _store_get(self) -> list: + return [DataPart(data=self.store.export())] + + async def _health(self, request: Request) -> Response: + waiting = [check.__name__ for check, _marker in _marked_members(self, _READY_ATTR) if not await _ready(check)] + if waiting: + return JSONResponse({"status": "starting", "waiting_for": waiting}, status_code=503) + return JSONResponse({"status": "ok"}) + + async def _session(self, request: Request) -> Response: + context = self.context(request) + if request.method == "DELETE": + context.sign_out() + elif context.session is None and context.user is not None: + context.sign_in(context.user) + return context.apply(JSONResponse({"user": context.user, "role": context.role, + "signed_in": context.user is not None})) + + async def _state(self, request: Request) -> Response: + return JSONResponse(self.store.state()) + + def _clock(self) -> UniverseClock: + if not hasattr(self, "clock"): + raise TypeError(f"{type(self).__qualname__}.__init__ must call super().__init__()") + return self.clock + + def _session_signer(self) -> SessionSigner: + if not hasattr(self, "_signer"): + raise TypeError(f"{type(self).__qualname__}.__init__ must call super().__init__()") + return self._signer + + @classmethod + def _site_path(cls, path: str) -> Path: + given = Path(path) + return given if given.is_absolute() else Path(inspect.getfile(cls)).resolve().parent / given + + +for _hook in _DEFAULT_HOOKS.values(): + setattr(getattr(WebsiteEnvironment, _hook), _DEFAULT_HOOK_ATTR, True) + + +async def _ready(check: Any) -> bool: + try: + return bool(await _maybe_await(check())) + except Exception: + return False + + +def _check_health(cls: type) -> None: + health = cls.declaration().health + clash = [r.name for r in routes_of(cls) if r.method == "GET" and openapi_path(r.path) == health] + if clash: + raise ValueError(f"{cls.__name__}'s health path {health} is served by @api {clash[0]}; use @ready_check") + + +class _Forwarded: + """Puts the address the browser reached the site at in the scope, from ``X-Forwarded-Host`` and + ``X-Forwarded-Proto``, before anything routes it, so redirects, ``request.url`` and a mounted app's ``url_for`` + name that address and not the internal one the proxy reached the server at.""" + + def __init__(self, app: Any) -> None: + self.app = app + + async def __call__(self, scope: dict, receive: Any, send: Any) -> None: + if scope["type"] in ("http", "websocket"): + headers = dict(scope["headers"]) + host = _first(headers.get(b"x-forwarded-host")) + proto = _first(headers.get(b"x-forwarded-proto")).lower() + scope = dict(scope) + if host: + scope["headers"] = [(k, v) for k, v in scope["headers"] if k != b"host"] + [(b"host", host.encode())] + if proto in ("http", "https"): + scope["scheme"] = proto if scope["type"] == "http" else {"http": "ws", "https": "wss"}[proto] + await self.app(scope, receive, send) + + +def _first(value: bytes | None) -> str: + return value.decode("latin-1").split(",")[0].strip() if value else "" + + +class _Universe: + """Stamps ``X-Universe-Now`` on every answer, and with ``clock_patch`` puts the clock script in its HTML.""" + + def __init__(self, app: Any, env: WebsiteEnvironment) -> None: + self.app, self.env = app, env + + async def __call__(self, scope: dict, receive: Any, send: Any) -> None: + if scope["type"] != "http": + return await self.app(scope, receive, send) + clock = self.env._clock() + await clock.refresh() + patch = self.env.declaration().clock_patch and scope["method"] != "HEAD" + held: dict = {} + + async def stamped(message: dict) -> None: + if message["type"] == "http.response.start": + headers = MutableHeaders(scope=message) + now = clock.now() + headers[UNIVERSE_NOW_HEADER] = iso(now) + if patch and headers.get("content-type", "").startswith("text/html") and "content-encoding" not in headers: + held.update(start=message, body=[], now=now) + return + elif held and message["type"] == "http.response.body": + held["body"].append(message.get("body", b"")) + if message.get("more_body"): + return + body = _with_script(b"".join(held["body"]), clock_patch_script(held["now"])) + MutableHeaders(scope=held["start"])["content-length"] = str(len(body)) + await send(held["start"]) + message = {"type": "http.response.body", "body": body} + await send(message) + + await self.app(scope, receive, stamped) + + +def _with_script(html: bytes, script: bytes) -> bytes: + head = _HEAD.search(html) + return html[:head.end()] + script + html[head.end():] if head else script + html + + +class _Fallback: + """What no route answers: a file of the static folder, then the mounted app, then for a page's path the folder's + ``index.html``; the data plane's and card's paths answer 404 however they're spelled.""" + + def __init__(self, root: Path | None, app: Any) -> None: + self.root = root.resolve() if root is not None else None + self.app = app + + async def __call__(self, scope: dict, receive: Any, send: Any) -> None: + if scope["type"] != "http": + if self.app is not None: + await self.app(scope, receive, send) + return + path, reading = scope["path"], scope["method"] in ("GET", "HEAD") + if reserved_path(path): + return await error_response(NotFound("not found"))(scope, receive, send) + if reading and (file := self._file(path)) is not None: + return await FileResponse(file)(scope, receive, send) + index = self._file("/") if reading and _page(path) else None + if self.app is not None: + return await self._app_or(index, scope, receive, send) + if index is not None: + return await FileResponse(index)(scope, receive, send) + await error_response(NotFound("not found"))(scope, receive, send) + + def _file(self, path: str) -> Path | None: + if self.root is None: + return None + candidate = (self.root / path.lstrip("/")).resolve() + if not candidate.is_relative_to(self.root): + return None + if candidate.is_dir(): + candidate = candidate / "index.html" + return candidate if candidate.is_file() else None + + async def _app_or(self, index: Path | None, scope: dict, receive: Any, send: Any) -> None: + """The mounted app's answer, or ``index`` instead of the app's 404 when there is one.""" + missed = False + + async def unless_missed(message: dict) -> None: + nonlocal missed + if message["type"] == "http.response.start" and message["status"] == 404 and index is not None: + missed = True + if not missed: + await send(message) + + await self.app(scope, receive, unless_missed) + if missed: + await FileResponse(index)(scope, receive, send) + + +def _page(path: str) -> bool: + """Whether ``path`` is a page a single-page app routes itself: not the API's, and not a file's.""" + return not (path == "/api" or path.startswith("/api/")) and "." not in path.rsplit("/", 1)[-1] diff --git a/packages/agentenv-protocol/src/agentenv_protocol/website/errors.py b/packages/agentenv-protocol/src/agentenv_protocol/website/errors.py new file mode 100644 index 00000000..13e349b8 --- /dev/null +++ b/packages/agentenv-protocol/src/agentenv_protocol/website/errors.py @@ -0,0 +1,39 @@ +"""Errors an ``@api`` method raises to answer with an HTTP status: the body is ``{"error": message, "code": code}``.""" +from __future__ import annotations + + +class ApiError(Exception): + """An answer other than success, ``status`` with ``{"error": message, "code": code}``; a subclass sets both + defaults, and either may be given per raise.""" + + status: int = 500 + code: str = "error" + + def __init__(self, message: str, *, code: str | None = None, status: int | None = None) -> None: + super().__init__(message) + self.message = message + self.code = code or type(self).code + self.status = status or type(self).status + + def body(self) -> dict: + return {"error": self.message, "code": self.code} + + +class BadRequest(ApiError): + status, code = 400, "bad_request" + + +class Unauthorized(ApiError): + status, code = 401, "unauthorized" + + +class Forbidden(ApiError): + status, code = 403, "forbidden" + + +class NotFound(ApiError): + status, code = 404, "not_found" + + +class Conflict(ApiError): + status, code = 409, "conflict" diff --git a/packages/agentenv-protocol/src/agentenv_protocol/website/routes.py b/packages/agentenv-protocol/src/agentenv_protocol/website/routes.py new file mode 100644 index 00000000..eb31a1fc --- /dev/null +++ b/packages/agentenv-protocol/src/agentenv_protocol/website/routes.py @@ -0,0 +1,350 @@ +"""A site's API: ``@api.get/post/put/patch/delete(path)`` methods, each bound from its type hints. A parameter named in +the path is a path parameter, one annotated with a pydantic model or ``dict`` is the JSON body, one annotated +``ApiContext`` receives the request's context, and every other one is a query parameter; path and query values are +coerced to their annotated types (int, float, bool, datetime, UUID, enum, Literal). What the method returns is the +JSON answer, a Starlette ``Response`` as it is; an ``ApiError`` it raises answers with its status, and a request its +parameters can't be bound from answers 400.""" +from __future__ import annotations + +import dataclasses +import datetime +import hmac +import inspect +import json +import logging +import os +import posixpath +import re +from typing import Any, Callable, Literal, NamedTuple, Optional, get_args, get_origin, get_type_hints + +from pydantic import BaseModel, TypeAdapter, ValidationError +from pydantic_core import to_jsonable_python +from starlette.requests import Request +from starlette.responses import JSONResponse, Response + +from ..agent_env_environment import _coerce_scalar, _maybe_await, _schema_for_annotation, _unwrap_annotation +from ..tool_context import DEFAULT_ROLE, caller_from_request +from ..types import RPC_PATH, WELL_KNOWN_PATH +from .errors import ApiError, BadRequest, Forbidden +from .session import SESSION_COOKIE, SessionSigner + +logger = logging.getLogger(__name__) + +_ROUTE_ATTR = "_aee_api_route" +_PATH_PARAM = re.compile(r"{(\w+)(?::\w+)?}") +Kind = Literal["path", "query", "body", "context"] + + +def reserved_path(path: str) -> bool: + """Whether ``path`` is the data plane's or the card's, however it's spelled, as the gateway decides it.""" + normalized = posixpath.normpath("/" + path.lstrip("/")) + return normalized == RPC_PATH or normalized.startswith(RPC_PATH + "/") or normalized == WELL_KNOWN_PATH + + +class ApiRoute(NamedTuple): + """One ``@api`` method: its verb, its Starlette path, its name on the card and in the OpenAPI document, the roles + it answers (any when empty), the status it succeeds with, and its function.""" + + method: str + path: str + name: str + roles: tuple[str, ...] + status: int + fn: Callable + + def advertised(self) -> dict: + return {"method": self.method, "path": openapi_path(self.path), "name": self.name, + "roles": list(self.roles) or None} + + +class _Api: + """``api.get(path, name=..., roles=[...])`` and its siblings mark a method as one route of the site's API.""" + + def get(self, path: str, **options: Any) -> Callable: + return self.route("GET", path, **options) + + def post(self, path: str, **options: Any) -> Callable: + return self.route("POST", path, **options) + + def put(self, path: str, **options: Any) -> Callable: + return self.route("PUT", path, **options) + + def patch(self, path: str, **options: Any) -> Callable: + return self.route("PATCH", path, **options) + + def delete(self, path: str, **options: Any) -> Callable: + return self.route("DELETE", path, **options) + + def route(self, method: str, path: str, *, name: str | None = None, roles: list[str] | None = None, + status: int = 200) -> Callable: + if not path.startswith("/"): + raise ValueError(f"an @api path starts with '/', not {path!r}") + if reserved_path(path): + raise ValueError(f"@api path {path!r} is under the data plane's {RPC_PATH} or the card's {WELL_KNOWN_PATH}") + + def mark(fn: Callable) -> Callable: + missing = sorted(set(_PATH_PARAM.findall(path)) - set(inspect.signature(fn).parameters)) + if missing: + raise TypeError(f"{fn.__qualname__} takes no parameter for the path's {missing}") + setattr(fn, _ROUTE_ATTR, ApiRoute(method.upper(), path, name or fn.__name__, tuple(roles or ()), status, fn)) + return fn + + return mark + + +api = _Api() + + +def routes_of(cls: type) -> list[ApiRoute]: + """The class's ``@api`` routes in the order they were first defined, each as the class resolves its name; + a name or a verb and path served twice is refused.""" + names = dict.fromkeys(name for klass in reversed(cls.__mro__) for name in vars(klass)) + routes = [route for name in names if (route := getattr(getattr(cls, name, None), _ROUTE_ATTR, None)) is not None] + for label, keys in (("name", [r.name for r in routes]), ("route", [f"{r.method} {r.path}" for r in routes])): + if twice := sorted({k for k in keys if keys.count(k) > 1}): + raise ValueError(f"{cls.__name__} serves the @api {label} {twice} twice") + return routes + + +def openapi_path(path: str) -> str: + return _PATH_PARAM.sub(r"{\1}", path) + + +@dataclasses.dataclass +class ApiContext: + """The request an ``@api`` method answers: its headers, cookies and query, the user it's signed in as (the + session cookie's, else the site's acting user), the caller's role (``request_role``), and the site's own address as the browser reached it. ``sign_in`` and ``sign_out`` set the + session cookie on the answer.""" + + request: Request + user: Optional[str] + role: str + session: Optional[dict] + _signer: SessionSigner = dataclasses.field(repr=False, default_factory=SessionSigner) + _set_cookie: list = dataclasses.field(repr=False, default_factory=list) + + @property + def headers(self) -> Any: + return self.request.headers + + @property + def cookies(self) -> dict[str, str]: + return self.request.cookies + + @property + def query(self) -> Any: + return self.request.query_params + + @property + def base_url(self) -> str: + return base_url(self.request) + + def sign_in(self, user: str) -> None: + self._set_session(user) + + def sign_out(self) -> None: + self._set_session(None) + + def _set_session(self, user: str | None) -> None: + value, expires = self._signer.encode(user) + self._set_cookie.append((value, expires)) + self.user, self.session = user, {"user": user} + + def apply(self, response: Response) -> Response: + for value, expires in self._set_cookie: + at = datetime.datetime.fromtimestamp(expires, datetime.timezone.utc) + response.set_cookie(SESSION_COOKIE, value, expires=at, path="/", httponly=True, samesite="lax") + return response + + +def base_url(request: Request) -> str: + """The site's address as the browser reached it: ``X-Forwarded-Proto`` and ``X-Forwarded-Host`` when a proxy + sent them, the request's own scheme and ``Host`` otherwise.""" + proto = request.headers.get("x-forwarded-proto", "").split(",")[0].strip() or request.url.scheme + host = (request.headers.get("x-forwarded-host", "").split(",")[0].strip() or request.headers.get("host") + or request.url.netloc) + return f"{proto}://{host}" + + +ROLE_KEY_ENV = "WEBSITE_ROLE_KEY" +ROLE_KEY_HEADER = "AgentEnv-Role-Key" + + +def request_role(request: Request) -> str: + """The role ``AgentEnv-Role`` names, believed only from a caller that also sends ``AgentEnv-Role-Key`` equal to + ``WEBSITE_ROLE_KEY``; every other caller, and every caller of a site with no key set, is ``default``. The gateway + forwards whatever role a browser sends, so without the key a page could claim any role.""" + claimed = caller_from_request(None, request.headers).role or DEFAULT_ROLE + key = os.environ.get(ROLE_KEY_ENV, "") + if claimed == DEFAULT_ROLE or not key: + return DEFAULT_ROLE + given = request.headers.get(ROLE_KEY_HEADER, "") + return claimed if hmac.compare_digest(given.encode(), key.encode()) else DEFAULT_ROLE + + +class _Param(NamedTuple): + name: str + kind: Kind + annotation: Any + default: Any + + +def plan(route: ApiRoute) -> list[_Param]: + """How each of the route's parameters is bound; at most one is the JSON body, and never on a GET.""" + hints = _hints(route.fn) + path_params = set(_PATH_PARAM.findall(route.path)) + params = [] + for name, param in list(inspect.signature(route.fn).parameters.items())[1:]: + if param.kind in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD): + continue + annotation = hints.get(name, param.annotation) + kind: Kind = ("context" if _is_context(annotation) else "path" if name in path_params + else "body" if _body_type(annotation) is not None else "query") + params.append(_Param(name, kind, annotation, param.default)) + bodies = [p.name for p in params if p.kind == "body"] + if len(bodies) > 1 or (bodies and route.method == "GET"): + raise TypeError(f"{route.fn.__qualname__}: a route takes at most one JSON body and a GET route none, " + f"not {bodies}") + return params + + +def _is_context(annotation: Any) -> bool: + if isinstance(annotation, str): + return annotation.strip("'\"").rsplit(".", 1)[-1] in ("ApiContext", "Optional[ApiContext]") + return any(t is ApiContext for t in _unwrap_annotation(annotation)) + + +def _body_type(annotation: Any) -> type | None: + for t in _unwrap_annotation(annotation): + base = get_origin(t) or t + if base is dict or (isinstance(base, type) and issubclass(base, BaseModel)): + return t + return None + + +async def bind(params: list[_Param], request: Request, context: ApiContext) -> dict: + """The method's arguments from ``request``; raises BadRequest for one that is missing or doesn't fit.""" + kwargs: dict = {} + for p in params: + if p.kind == "context": + kwargs[p.name] = context + elif p.kind == "body": + kwargs[p.name] = await _body(p, request) + else: + source = request.path_params if p.kind == "path" else request.query_params + if p.name not in source: + if p.default is inspect.Parameter.empty: + raise BadRequest(f"missing {p.kind} parameter {p.name!r}", code="invalid_params") + kwargs[p.name] = p.default + continue + many = _list_type(p.annotation) if p.kind == "query" else None + try: + kwargs[p.name] = ([_coerce_scalar(v, get_args(many)[0] if get_args(many) else None) + for v in source.getlist(p.name)] if many is not None + else _coerce_scalar(source[p.name], p.annotation)) + except (TypeError, ValueError) as e: + raise BadRequest(f"{p.kind} parameter {p.name!r}: {e}", code="invalid_params") from None + return kwargs + + +def _list_type(annotation: Any) -> Any: + """The ``list[T]`` an annotation names, Optional or Annotated or not; None when it names no list.""" + inner = _unwrap_annotation(annotation) + return inner[0] if len(inner) == 1 and (inner[0] is list or get_origin(inner[0]) is list) else None + + +async def _body(p: _Param, request: Request) -> Any: + raw = await request.body() + if not raw: + if p.default is not inspect.Parameter.empty: + return p.default + raise BadRequest("this route takes a JSON body", code="invalid_params") + try: + value = json.loads(raw) + except ValueError: + raise BadRequest("the request body is not JSON", code="invalid_params") from None + try: + return TypeAdapter(_body_type(p.annotation)).validate_python(value) + except ValidationError as e: + problems = "; ".join(f"{'.'.join(map(str, err['loc'])) or 'body'}: {err['msg']}" for err in e.errors()) + raise BadRequest(f"invalid body: {problems}", code="invalid_params") from None + + +def endpoint(fn: Callable, route: ApiRoute, context_for: Callable[[Request], ApiContext]) -> Callable: + """The Starlette endpoint of one route, calling the bound method ``fn``.""" + params = plan(route) + + async def handle(request: Request) -> Response: + context = None + try: + context = context_for(request) + if route.roles and context.role not in route.roles: + raise Forbidden(f"role {context.role!r} may not call {route.name}") + result = await _maybe_await(fn(**await bind(params, request, context))) + if isinstance(result, Response): + response = result + elif not _has_body(route.status): + response = Response(status_code=route.status) + else: + response = JSONResponse(to_jsonable_python(result), status_code=route.status) + except ApiError as e: + response = error_response(e) + except Exception: + logger.exception("@api %s failed", route.name) + response = error_response(ApiError("internal error", code="internal")) + return context.apply(response) if context is not None else response + + return handle + + +def _has_body(status: int) -> bool: + """Whether an answer with ``status`` may carry a body: not a 1xx, 204 or 304.""" + return status >= 200 and status not in (204, 304) + + +def error_response(error: ApiError) -> Response: + return JSONResponse(error.body(), status_code=error.status) + + +def openapi(title: str, routes: list[ApiRoute]) -> dict: + """The OpenAPI 3.1 document of ``routes``, each operation's id its route's name.""" + paths: dict = {} + for route in routes: + doc = inspect.getdoc(route.fn) or "" + operation: dict = {"operationId": route.name, "parameters": [], + "responses": {str(route.status): {"description": "Success"}}} + if doc: + operation["summary"], operation["description"] = doc.splitlines()[0], doc + if route.roles: + operation["x-agentenv-roles"] = list(route.roles) + for p in plan(route): + if p.kind in ("path", "query"): + schema = _schema_for_annotation(p.annotation) + if p.default is not inspect.Parameter.empty: + schema["default"] = to_jsonable_python(p.default) + operation["parameters"].append({"name": p.name, "in": p.kind, "schema": schema, + "required": p.kind == "path" or p.default is inspect.Parameter.empty}) + elif p.kind == "body": + operation["requestBody"] = {"required": p.default is inspect.Parameter.empty, + "content": {"application/json": {"schema": _json_schema(_body_type(p.annotation))}}} + returns = _json_schema(_hints(route.fn).get("return")) + if returns: + operation["responses"][str(route.status)]["content"] = {"application/json": {"schema": returns}} + paths.setdefault(openapi_path(route.path), {})[route.method.lower()] = operation + return {"openapi": "3.1.0", "info": {"title": title, "version": "1.0"}, "paths": paths} + + +def _hints(fn: Callable) -> dict: + try: + return get_type_hints(fn) + except Exception: + return {} + + +def _json_schema(annotation: Any) -> dict: + if annotation is None or annotation is type(None) or annotation is inspect.Parameter.empty: + return {} + try: + return TypeAdapter(annotation).json_schema() + except Exception: + return {} diff --git a/packages/agentenv-protocol/src/agentenv_protocol/website/session.py b/packages/agentenv-protocol/src/agentenv_protocol/website/session.py new file mode 100644 index 00000000..8ee9b8e7 --- /dev/null +++ b/packages/agentenv-protocol/src/agentenv_protocol/website/session.py @@ -0,0 +1,59 @@ +"""A site's signed session cookie: who the browser is signed in as, or that it signed out, signed with HMAC-SHA256 +under ``WEBSITE_SESSION_SECRET``, or a secret of the process's own when that isn't set. The cookie names no domain, +so it holds under whatever host the site is reached at.""" +from __future__ import annotations + +import base64 +import binascii +import datetime +import hashlib +import hmac +import json +import os +import secrets +import time + +SESSION_COOKIE = "agentenv_session" +SECRET_ENV = "WEBSITE_SESSION_SECRET" +SESSION_TTL_S = 7 * 24 * 3600 + + +class SessionSigner: + def __init__(self, secret: str | bytes | None = None) -> None: + secret = secret if secret is not None else os.environ.get(SECRET_ENV) + self._key = (secret.encode() if isinstance(secret, str) else secret) or secrets.token_bytes(32) + + def encode(self, user: str | None, ttl_s: int = SESSION_TTL_S) -> tuple[str, int]: + """The cookie value for ``user`` signed in, or signed out when None, and when it expires, in epoch + seconds.""" + expires = int(time.time()) + ttl_s + payload = base64.urlsafe_b64encode(json.dumps({"u": user, "exp": expires}).encode()).decode().rstrip("=") + return f"{payload}.{self._sign(payload)}", expires + + def decode(self, value: str | None) -> dict | None: + """The session a cookie value carries, ``{"user": ...}``; None for one that is absent, forged or expired.""" + payload, _, signature = (value or "").partition(".") + if not payload or not hmac.compare_digest(signature.encode("utf-8", "surrogateescape"), + self._sign(payload).encode()): + return None + try: + data = json.loads(base64.urlsafe_b64decode(payload + "=" * (-len(payload) % 4))) + except (ValueError, binascii.Error): + return None + if not isinstance(data, dict) or not isinstance(data.get("exp"), int) or data["exp"] < time.time(): + return None + return {"user": data.get("u")} + + def cookie(self, user: str | None, ttl_s: int = SESSION_TTL_S) -> dict: + """The cookie a browser sets to be signed in as ``user``, as a browser automation adds one; its domain is + left to whoever adds it.""" + value, expires = self.encode(user, ttl_s) + return {"name": SESSION_COOKIE, "value": value, "path": "/", "expires": expires, "httpOnly": True, + "sameSite": "Lax"} + + def _sign(self, payload: str) -> str: + return hmac.new(self._key, payload.encode("utf-8", "surrogateescape"), hashlib.sha256).hexdigest() + + +def expires_at(epoch: int) -> str: + return datetime.datetime.fromtimestamp(epoch, datetime.timezone.utc).isoformat().replace("+00:00", "Z") diff --git a/packages/agentenv-protocol/src/agentenv_protocol/website/store.py b/packages/agentenv-protocol/src/agentenv_protocol/website/store.py new file mode 100644 index 00000000..ce21ad84 --- /dev/null +++ b/packages/agentenv-protocol/src/agentenv_protocol/website/store.py @@ -0,0 +1,134 @@ +"""The default state store of a simple site: named collections of JSON objects, each with an ``id``. Its data is a JSON +object of collections by name, beside which a string ``current_user`` names the user a browser starts signed in as. +It remembers the state it was last loaded with, so a verifier can read the initial state, the current one and the +difference between them.""" +from __future__ import annotations + +import base64 +import copy +import json +from pathlib import Path +from typing import Any, Iterable +from urllib.parse import unquote, urlparse + +from ..types import DataPart, FilePart, Part, TextPart + +CURRENT_USER = "current_user" + + +class Collections: + """Named lists of JSON objects; ``store["orders"]`` is the list itself, created empty on first use.""" + + def __init__(self, data: dict | None = None) -> None: + self._collections: dict[str, list[dict]] = {} + self.current_user: str | None = None + self.initial: dict = {} + if data: + self.load(data) + self.mark_initial() + + def __getitem__(self, name: str) -> list[dict]: + return self._collections.setdefault(name, []) + + def names(self) -> list[str]: + return list(self._collections) + + def get(self, name: str, id: Any) -> dict | None: + return next((doc for doc in self[name] if doc.get("id") == id), None) + + def find(self, name: str, **match: Any) -> list[dict]: + return [doc for doc in self[name] if all(doc.get(k) == v for k, v in match.items())] + + def insert(self, name: str, doc: dict) -> dict: + """Add ``doc``, giving it the next integer ``id`` of its collection when it has none.""" + doc = dict(doc) + if doc.get("id") is None: + doc["id"] = 1 + max((d["id"] for d in self[name] if isinstance(d.get("id"), int)), default=0) + self[name].append(doc) + return doc + + def update(self, name: str, id: Any, **fields: Any) -> dict | None: + doc = self.get(name, id) + if doc is not None: + doc.update(fields) + return doc + + def delete(self, name: str, id: Any) -> bool: + doc = self.get(name, id) + if doc is not None: + self[name].remove(doc) + return doc is not None + + def load(self, data: dict) -> None: + """Add ``data``'s collections to the store's, and take its ``current_user`` when it names one.""" + if not isinstance(data, dict): + raise ValueError("a store's data is a JSON object of collections by name") + for name, docs in data.items(): + if name == CURRENT_USER: + self.current_user = docs + elif isinstance(docs, list) and all(isinstance(d, dict) for d in docs): + self[name].extend(copy.deepcopy(docs)) + else: + raise ValueError(f"collection {name!r} is not a list of JSON objects") + + def load_parts(self, parts: Iterable[Part]) -> None: + for part in parts: + self.load(_part_data(part)) + + def reset(self) -> None: + self._collections.clear() + self.current_user = None + self.initial = {} + + def export(self) -> dict: + data: dict = copy.deepcopy(self._collections) + if self.current_user is not None: + data[CURRENT_USER] = self.current_user + return data + + def mark_initial(self) -> None: + self.initial = self.export() + + def diff(self) -> dict: + """Each collection that changed since the initial state, as its ``added``, ``removed`` and ``changed`` + objects, matched by ``id``; a changed one as its ``id``, ``before`` and ``after``.""" + current = self.export() + changes: dict = {} + for name in dict.fromkeys([*self.initial, *current]): + if name == CURRENT_USER: + continue + before = {_key(d): d for d in self.initial.get(name, [])} + after = {_key(d): d for d in current.get(name, [])} + entry = { + "added": [d for k, d in after.items() if k not in before], + "removed": [d for k, d in before.items() if k not in after], + "changed": [{"id": d.get("id"), "before": before[k], "after": d} + for k, d in after.items() if k in before and before[k] != d], + } + if any(entry.values()): + changes[name] = entry + return changes + + def state(self) -> dict: + return {"initial_state": copy.deepcopy(self.initial), "current_state": self.export(), "state_diff": self.diff()} + + +def _key(doc: dict) -> str: + return json.dumps(doc.get("id")) if doc.get("id") is not None else json.dumps(doc, sort_keys=True) + + +def _part_data(part: Part) -> dict: + """A part's JSON: a data part's data, a text part's text, or a file part's bytes or ``file://`` file.""" + if isinstance(part, DataPart): + return part.data + if isinstance(part, TextPart): + return json.loads(part.text) + if isinstance(part, FilePart): + file = part.file + if getattr(file, "bytes", None) is not None: + return json.loads(base64.b64decode(file.bytes)) + uri = urlparse(file.uri) + if uri.scheme == "file": + return json.loads(Path(unquote(uri.path)).read_text(encoding="utf-8")) + raise ValueError(f"the store reads file parts by their bytes or a file:// URI, not {file.uri!r}") + raise ValueError(f"the store can't read a {type(part).__name__}") diff --git a/packages/agentenv-protocol/tests/test_website_api.py b/packages/agentenv-protocol/tests/test_website_api.py new file mode 100644 index 00000000..04d96744 --- /dev/null +++ b/packages/agentenv-protocol/tests/test_website_api.py @@ -0,0 +1,308 @@ +"""The website SDK's @api routes: binding from type hints, JSON answers, errors, roles, and what a class may not +declare.""" + +from __future__ import annotations + +import datetime +import enum +from typing import Optional + +import pytest +from pydantic import BaseModel +from starlette.responses import PlainTextResponse +from starlette.testclient import TestClient + +from agentenv_protocol.website import ( + ROLE_KEY_ENV, + ROLE_KEY_HEADER, + ApiContext, + BadRequest, + Conflict, + NotFound, + WebsiteEnvironment, + api, + reserved_path, + website, +) + + +class Color(enum.Enum): + RED = "red" + BLUE = "blue" + + +class NewItem(BaseModel): + title: str + price: float = 0 + + +@website(domains=["shop.example"]) +class Shop(WebsiteEnvironment): + def __init__(self) -> None: + super().__init__() + self.items: dict[int, dict] = {1: {"id": 1, "title": "mug", "price": 3.5}} + + @api.get("/api/items/{item_id}", name="get_item") + async def get_item(self, item_id: int) -> dict: + """Read one item.""" + if item_id not in self.items: + raise NotFound(f"no item {item_id}") + return self.items[item_id] + + @api.get("/api/items") + def list_items(self, limit: int = 10, cheap: bool = False, color: Optional[Color] = None, + since: Optional[datetime.date] = None, tags: list[str] = []) -> dict: + return {"limit": limit, "cheap": cheap, "color": color.value if color else None, + "since": since.isoformat() if since else None, "tags": tags} + + @api.post("/api/items", name="create_item", status=201) + async def create_item(self, item: NewItem) -> NewItem: + if any(i["title"] == item.title for i in self.items.values()): + raise Conflict(f"{item.title} exists") + self.items[len(self.items) + 1] = {"id": len(self.items) + 1, **item.model_dump()} + return item + + @api.patch("/api/items/{item_id}") + async def patch_item(self, item_id: int, changes: dict) -> dict: + self.items[item_id].update(changes) + return self.items[item_id] + + @api.delete("/api/items/{item_id}", roles=["admin"]) + async def delete_item(self, item_id: int) -> None: + self.items.pop(item_id, None) + + @api.put("/api/teapot") + async def teapot(self) -> None: + raise BadRequest("short and stout", code="teapot", status=418) + + @api.get("/api/whoami") + async def whoami(self, ctx: ApiContext) -> dict: + return {"user": ctx.user, "role": ctx.role, "base_url": ctx.base_url, "q": ctx.query.get("q"), + "header": ctx.headers.get("x-probe"), "cookie": ctx.cookies.get("c")} + + @api.get("/api/plain") + async def plain(self) -> PlainTextResponse: + return PlainTextResponse("as is") + + @api.get("/api/boom") + async def boom(self) -> dict: + raise RuntimeError("secret detail") + + @api.get("/api/picked") + def picked(self, ids: Optional[list[int]] = None) -> dict: + return {"ids": ids} + + @api.delete("/api/cart", status=204) + async def empty_cart(self) -> None: + return None + + +@pytest.fixture +def client() -> TestClient: + return TestClient(Shop().create_app()) + + +def test_path_query_and_body_parameters_are_bound_from_type_hints(client: TestClient) -> None: + assert client.get("/api/items/1").json() == {"id": 1, "title": "mug", "price": 3.5} + assert client.get("/api/items").json() == {"limit": 10, "cheap": False, "color": None, "since": None, "tags": []} + listed = client.get("/api/items?limit=3&cheap=yes&color=blue&since=2030-01-02&tags=a&tags=b").json() + assert listed == {"limit": 3, "cheap": True, "color": "blue", "since": "2030-01-02", "tags": ["a", "b"]} + created = client.post("/api/items", json={"title": "pen", "price": 1}) + assert created.status_code == 201 and created.json() == {"title": "pen", "price": 1.0} + assert client.patch("/api/items/1", json={"price": 4}).json()["price"] == 4 + + +def test_a_request_its_parameters_cannot_be_bound_from_answers_400(client: TestClient) -> None: + for response in (client.get("/api/items/abc"), client.get("/api/items?limit=many"), + client.get("/api/items?color=green"), client.post("/api/items", json={"price": 2}), + client.post("/api/items", content=b"not json"), client.post("/api/items"), + client.patch("/api/items/1", json=[1, 2])): + assert response.status_code == 400, response.text + assert response.json()["code"] == "invalid_params" + assert "title" in client.post("/api/items", json={"price": 2}).json()["error"] + + +def test_api_errors_answer_with_their_status_and_code(client: TestClient) -> None: + missing = client.get("/api/items/9") + assert (missing.status_code, missing.json()) == (404, {"error": "no item 9", "code": "not_found"}) + assert client.post("/api/items", json={"title": "mug"}).json() == {"error": "mug exists", "code": "conflict"} + teapot = client.put("/api/teapot") + assert (teapot.status_code, teapot.json()["code"]) == (418, "teapot") + boom = client.get("/api/boom") + assert (boom.status_code, boom.json()) == (500, {"error": "internal error", "code": "internal"}) + + +def test_responses_pass_through_and_none_is_json_null(client: TestClient, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv(ROLE_KEY_ENV, "k3y") + assert client.get("/api/plain").text == "as is" + deleted = client.delete("/api/items/1", headers={"AgentEnv-Role": "admin", ROLE_KEY_HEADER: "k3y"}) + assert (deleted.status_code, deleted.json()) == (200, None) + + +def test_an_optional_list_query_parameter_is_a_list_of_its_coerced_values(client: TestClient) -> None: + assert client.get("/api/picked?ids=1&ids=2").json() == {"ids": [1, 2]} + assert client.get("/api/picked").json() == {"ids": None} + assert client.get("/api/picked?ids=x").status_code == 400 + + +def test_a_route_declared_204_answers_with_no_body(client: TestClient) -> None: + emptied = client.delete("/api/cart") + assert (emptied.status_code, emptied.content) == (204, b"") + assert "content-length" not in emptied.headers and "content-type" not in emptied.headers + + +def test_a_204_answer_keeps_a_real_server_connection_usable() -> None: + import socket + import threading + import time + + uvicorn = pytest.importorskip("uvicorn") + server = uvicorn.Server(uvicorn.Config(Shop().create_app(), host="127.0.0.1", port=0, log_level="error", + ws="none")) + thread = threading.Thread(target=server.run, daemon=True) + thread.start() + while not server.started: + time.sleep(0.01) + port = server.servers[0].sockets[0].getsockname()[1] + try: + with socket.create_connection(("127.0.0.1", port), timeout=5) as conn: + conn.sendall(b"DELETE /api/cart HTTP/1.1\r\nHost: x\r\n\r\nGET /api/items/1 HTTP/1.1\r\nHost: x\r\n\r\n") + received = b"" + while b"mug" not in received: + chunk = conn.recv(65536) + if not chunk: + break + received += chunk + assert received.startswith(b"HTTP/1.1 204") and b"HTTP/1.1 200" in received and b"mug" in received + finally: + server.should_exit = True + thread.join(5) + + +def test_a_session_cookie_that_is_not_ascii_is_no_session(client: TestClient) -> None: + for path in ("/api/whoami", "/api/session"): + response = client.get(path, headers={"Cookie": "agentenv_session=abc.\xe9".encode("latin-1")}) + assert response.status_code == 200, response.text + assert response.json()["user"] is None + + +def test_roles_refuse_other_callers_with_403(client: TestClient, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv(ROLE_KEY_ENV, "k3y") + for headers in ({}, {"AgentEnv-Role": "default"}, {"AgentEnv-Role": "agent", ROLE_KEY_HEADER: "k3y"}): + refused = client.delete("/api/items/1", headers=headers) + assert refused.status_code == 403 and refused.json()["code"] == "forbidden" + assert client.get("/api/items/1").status_code == 200 + assert client.delete("/api/items/1", headers={"AgentEnv-Role": "admin", ROLE_KEY_HEADER: "k3y"}).status_code == 200 + + +def test_a_role_is_believed_only_with_the_site_s_role_key(client: TestClient, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv(ROLE_KEY_ENV, raising=False) + claim = {"AgentEnv-Role": "admin"} + assert client.delete("/api/items/1", headers=claim).status_code == 403 + assert client.delete("/api/items/1", headers={**claim, ROLE_KEY_HEADER: ""}).status_code == 403 + monkeypatch.setenv(ROLE_KEY_ENV, "k3y") + for key in (None, "", "wrong", "k3y "): + headers = claim if key is None else {**claim, ROLE_KEY_HEADER: key} + assert client.delete("/api/items/1", headers=headers).status_code == 403 + assert client.get("/api/whoami", headers=headers).json()["role"] == "default" + assert client.get("/api/whoami", headers={**claim, ROLE_KEY_HEADER: "k3y"}).json()["role"] == "admin" + + +def test_the_context_carries_the_request_role_and_forwarded_address(client: TestClient, + monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv(ROLE_KEY_ENV, "k3y") + client.cookies.set("c", "crumb") + seen = client.get("/api/whoami?q=1", headers={"x-probe": "p", "AgentEnv-Role": "agent", ROLE_KEY_HEADER: "k3y"}).json() + assert seen == {"user": None, "role": "agent", "base_url": "http://testserver", "q": "1", "header": "p", + "cookie": "crumb"} + forwarded = client.get("/api/whoami", headers={"X-Forwarded-Host": "shop.example, proxy", + "X-Forwarded-Proto": "https"}).json() + assert (forwarded["base_url"], forwarded["role"]) == ("https://shop.example", "default") + + +def test_a_redirect_names_the_forwarded_address(client: TestClient) -> None: + redirected = client.get("/api/items/", follow_redirects=False, + headers={"Host": "internal-x.invalid", "X-Forwarded-Host": "shop.example", + "X-Forwarded-Proto": "https"}) + assert (redirected.status_code, redirected.headers["location"]) == (307, "https://shop.example/api/items") + assert client.get("/api/items/", follow_redirects=False).headers["location"] == "http://testserver/api/items" + + +def test_parameterized_routes_never_take_the_data_plane_or_card() -> None: + @website(domains=["wiki.example"]) + class Wiki(WebsiteEnvironment): + @api.get("/{slug}") + def page(self, slug: str) -> dict: + return {"slug": slug} + + @api.post("/{collection}") + def add(self, collection: str) -> dict: + return {"captured": collection} + + @api.get("/.well-known/{name}") + def well_known(self, name: str) -> dict: + return {"captured": name} + + @api.get("/files/{rest:path}") + def files(self, rest: str) -> dict: + return {"rest": rest} + + client = TestClient(Wiki().create_app()) + assert client.get("/hello").json() == {"slug": "hello"} + assert client.post("/notes").json() == {"captured": "notes"} + assert client.get("/.well-known/other.json").json() == {"captured": "other.json"} + assert client.get("/files/a/b.txt").json() == {"rest": "a/b.txt"} + reset = client.post("/agentenv", json={"jsonrpc": "2.0", "id": 1, "method": "data/get", "params": {}}).json() + assert reset["id"] == 1 and "result" in reset + assert client.get("/.well-known/agent-env.json").json()["name"] == "Wiki" + assert client.get("/api/health").json() == {"status": "ok"} + + +@pytest.mark.parametrize("path", ["/agentenv", "/agentenv/ext/x", "/.well-known/agent-env.json", + "/api/../agentenv/x", "//agentenv"]) +def test_api_paths_under_the_reserved_ones_are_refused(path: str) -> None: + assert reserved_path(path) + with pytest.raises(ValueError, match="data plane"): + api.get(path) + + +def test_a_class_serving_a_name_or_route_twice_is_refused() -> None: + with pytest.raises(ValueError, match="name"): + class Twice(WebsiteEnvironment): + @api.get("/a", name="same") + def a(self) -> None: ... + + @api.get("/b", name="same") + def b(self) -> None: ... + + with pytest.raises(ValueError, match="route"): + class Again(WebsiteEnvironment): + @api.post("/a") + def a(self) -> None: ... + + @api.post("/a") + def b(self) -> None: ... + + +def test_a_route_must_take_its_path_parameters_and_at_most_one_body() -> None: + with pytest.raises(TypeError, match="item_id"): + api.get("/api/{item_id}")(lambda self: None) + + class Bodies(WebsiteEnvironment): + @api.post("/api/x") + def x(self, a: dict, b: NewItem) -> None: ... + + with pytest.raises(TypeError, match="one JSON body"): + Bodies().create_app() + + +def test_the_health_path_is_not_an_api_route() -> None: + with pytest.raises(ValueError, match="ready_check"): + @website(health="/api/up") + class Clash(WebsiteEnvironment): + @api.get("/api/up") + def up(self) -> dict: + return {} + + with pytest.raises(ValueError, match="health"): + website(health="/agentenv/health") diff --git a/packages/agentenv-protocol/tests/test_website_conformance.py b/packages/agentenv-protocol/tests/test_website_conformance.py new file mode 100644 index 00000000..2a85d3dd --- /dev/null +++ b/packages/agentenv-protocol/tests/test_website_conformance.py @@ -0,0 +1,152 @@ +"""The website/v1 conformance checker against an SDK site, a hand-written server that meets the contract, and servers +that don't.""" + +from __future__ import annotations + +import datetime +from pathlib import Path + +import pytest +from starlette.applications import Starlette +from starlette.requests import Request +from starlette.responses import HTMLResponse, JSONResponse, Response +from starlette.routing import Route +from starlette.testclient import TestClient + +from agentenv_protocol.website import WebsiteEnvironment, api, website +from agentenv_protocol.website import conformance +from agentenv_protocol.website.clock import UNIVERSE_NOW_HEADER, iso + +CHECKS = ["card", "any_host", "health", "universe_now", "data_plane", "clock", "forwarded", "cookies"] + + +def statuses(results: list[conformance.Result]) -> dict[str, str]: + return {r.name: r.status for r in results} + + +@pytest.fixture +def sdk_site(tmp_path: Path) -> TestClient: + (tmp_path / "index.html").write_text("hi") + + @website(domains=["notes.example"], static=str(tmp_path)) + class Notes(WebsiteEnvironment): + @api.get("/api/notes") + def notes(self) -> list: + return self.store["notes"] + + env = Notes() + env.store.load({"current_user": "ann", "notes": [{"id": 1, "text": "hello"}]}) + return TestClient(env.create_app()) + + +def test_an_sdk_site_passes_every_check(sdk_site: TestClient) -> None: + results = conformance.check("http://testserver", client=sdk_site) + assert [r.name for r in results] == CHECKS + assert all(r.status == "passed" for r in results), results + assert sdk_site.post("/agentenv", json={"jsonrpc": "2.0", "id": 1, "method": "data/get"}).json()["result"][ + "parts"][0]["data"] == {"current_user": "ann", "notes": [{"id": 1, "text": "hello"}]} + + +def _handwritten(*, leak_host: bool = False, domain_cookie: bool = False, clock: bool = True, + leak_redirect: bool = False) -> Starlette: + """A server written without the SDK: the contract by hand, optionally broken.""" + state = {"items": ["a"], "url": None, "at": None} + + def now() -> datetime.datetime: + return state["at"] or datetime.datetime.now(datetime.timezone.utc) + + def stamp(response: Response) -> Response: + response.headers[UNIVERSE_NOW_HEADER] = iso(now()) + return response + + async def page(request: Request) -> Response: + host = request.headers["host"] if leak_host else request.headers.get("x-forwarded-host", request.headers["host"]) + response = stamp(HTMLResponse(f'x')) + response.set_cookie("s", "1", domain="example.com" if domain_cookie else None) + return response + + async def health(request: Request) -> Response: + return stamp(JSONResponse({"ok": True})) + + async def items(request: Request) -> Response: + return stamp(JSONResponse(state["items"])) + + async def items_slash(request: Request) -> Response: + host = request.headers["host"] if leak_redirect else request.headers.get("x-forwarded-host", request.headers["host"]) + return stamp(Response(status_code=307, headers={"Location": f"https://{host}/items"})) + + async def card(request: Request) -> Response: + extensions = [{"uri": "urn:agentenv:website/v1", + "params": {"domains": ["h.example"], "health": "/healthz", + "routes": [{"method": "GET", "path": "/items", "name": "items"}]}}] + if clock: + extensions.append({"uri": "urn:agentenv:clock/v1", "params": { + "endpoint": "/agentenv/ext/sync_time", "methods": {"sync_time": {"method": "POST"}}}}) + return stamp(JSONResponse({"name": "hand", "capabilities": {"extensions": extensions, + "operations": ["data/reset", "data/add", "data/get"]}})) + + async def rpc(request: Request) -> Response: + body = await request.json() + method = body["method"] + if method == "data/reset": + state["items"] = [] + elif method == "data/add": + for part in body["params"]["parts"]: + state["items"] += part["data"]["items"] + result = {"parts": [{"kind": "data", "data": {"items": state["items"]}}]} if method == "data/get" else {} + return stamp(JSONResponse({"jsonrpc": "2.0", "id": body["id"], "result": result})) + + async def sync_time(request: Request) -> Response: + import httpx + + url = (await request.json())["env_get_time_url"] + async with httpx.AsyncClient() as client: + virtual = (await client.get(url)).json()["virtual_time"] + state["at"] = datetime.datetime.fromisoformat(virtual.replace("Z", "+00:00")) + return stamp(JSONResponse({"gateway_virtual_time": virtual})) + + return Starlette(routes=[Route("/", page), Route("/healthz", health), Route("/items", items), + Route("/items/", items_slash), + Route("/.well-known/agent-env.json", card), Route("/agentenv", rpc, methods=["POST"]), + Route("/agentenv/ext/sync_time", sync_time, methods=["POST"])]) + + +def test_a_handwritten_server_meeting_the_contract_passes() -> None: + results = conformance.check("http://testserver", client=TestClient(_handwritten())) + assert statuses(results) == dict.fromkeys(CHECKS, "passed"), results + + +def test_a_server_breaking_the_contract_fails_those_checks() -> None: + results = statuses(conformance.check("http://testserver", + client=TestClient(_handwritten(leak_host=True, domain_cookie=True, clock=False)))) + assert results["forwarded"] == results["cookies"] == results["clock"] == "failed" + assert results["card"] == results["health"] == results["data_plane"] == "passed" + + +def test_the_forwarded_check_follows_no_redirect_and_reads_its_location(sdk_site: TestClient) -> None: + forwarded = {r.name: r for r in conformance.check("http://testserver", client=sdk_site)}["forwarded"] + assert forwarded.status == "passed" and "/api/notes/" in forwarded.detail + leaked = {r.name: r for r in conformance.check("http://testserver", + client=TestClient(_handwritten(leak_redirect=True)))}["forwarded"] + assert leaked.status == "failed" and "Location of /items/" in leaked.detail + + +def test_a_server_with_no_card_health_or_header_fails_them() -> None: + bare = Starlette(routes=[Route("/", lambda request: JSONResponse({"not": "html"}))]) + results = statuses(conformance.check("http://testserver", client=TestClient(bare))) + assert results["card"] == results["any_host"] == results["health"] == results["universe_now"] == "failed" + assert results["data_plane"] == results["clock"] == "failed" + assert results["cookies"] == "skipped" + + +def test_main_prints_each_check_and_exits_by_outcome(monkeypatch, capsys) -> None: + monkeypatch.setattr(conformance, "check", lambda url, **options: [conformance.Result("card", "passed"), + conformance.Result("clock", "failed", "late")]) + assert conformance.main(["http://site"]) == 1 + printed = capsys.readouterr().out + assert "PASS card" in printed and "FAIL clock: late" in printed + + +def test_main_exits_2_when_the_server_cannot_be_reached(capsys) -> None: + assert conformance.main(["http://127.0.0.1:9", "--timeout", "2"]) == 2 + assert "couldn't reach" in capsys.readouterr().err diff --git a/packages/agentenv-protocol/tests/test_website_environment.py b/packages/agentenv-protocol/tests/test_website_environment.py new file mode 100644 index 00000000..539e41f6 --- /dev/null +++ b/packages/agentenv-protocol/tests/test_website_environment.py @@ -0,0 +1,418 @@ +"""WebsiteEnvironment as a site's server: static pages, health, the card and its OpenAPI document, the default store +and its data plane, sessions, the universe's clock, and a mounted app.""" + +from __future__ import annotations + +import asyncio +import base64 +import datetime +import json +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + +import pytest +from starlette.applications import Starlette +from starlette.responses import PlainTextResponse +from starlette.routing import Route +from starlette.testclient import TestClient + +from agentenv_protocol import DataPart, get_data +from agentenv_protocol.website import ( + SESSION_COOKIE, + UNIVERSE_NOW_HEADER, + WEBSITE_EXTENSION_URI, + ApiContext, + Collections, + WebsiteEnvironment, + api, + ready_check, + website, +) +from agentenv_protocol.website import clock as clock_module +from agentenv_protocol.website.clock import parse_time + +RPC = "/agentenv" + + +def rpc(client: TestClient, method: str, params: dict | None = None) -> dict: + body = client.post(RPC, json={"jsonrpc": "2.0", "id": 1, "method": method, "params": params or {}}).json() + assert "error" not in body, body + return body["result"] + + +@pytest.fixture +def dist(tmp_path: Path) -> Path: + root = tmp_path / "dist" + (root / "assets").mkdir(parents=True) + (root / "index.html").write_text("Shophome") + (root / "assets" / "app.js").write_text("console.log(1)") + (tmp_path / "secret.txt").write_text("outside") + return root + + +def site(dist: Path, **declared) -> type[WebsiteEnvironment]: + @website(domains=["shop.example", "www.shop.example"], static=str(dist), **declared) + class Shop(WebsiteEnvironment): + @api.get("/api/orders", name="list_orders") + def list_orders(self, ctx: ApiContext) -> list: + """List the signed-in user's orders.""" + return self.store.find("orders", user=ctx.user) + + @api.post("/api/orders", name="place_order", roles=["default", "agent"]) + def place_order(self, order: dict, ctx: ApiContext) -> dict: + return self.store.insert("orders", {**order, "user": ctx.user, "at": self.now().isoformat()}) + + @api.post("/api/sign-in") + def sign_in(self, body: dict, ctx: ApiContext) -> dict: + ctx.sign_in(body["user"]) + return {"user": ctx.user} + + return Shop + + +SEED = {"current_user": "alice", "users": [{"id": "alice"}, {"id": "bob"}], "orders": [{"id": 1, "user": "bob"}]} + + +@pytest.fixture +def shop(dist: Path) -> WebsiteEnvironment: + env = site(dist)() + env.store.load(SEED) + env.store.mark_initial() + return env + + +@pytest.fixture +def client(shop: WebsiteEnvironment) -> TestClient: + return TestClient(shop.create_app()) + + +def test_static_files_and_index_fallback_answer_for_any_host(client: TestClient) -> None: + for host in ("testserver", "shop.example", "10.0.0.7:8000", "anything.invalid"): + page = client.get("/", headers={"Host": host}) + assert page.status_code == 200 and "home" in page.text and page.headers["content-type"].startswith("text/html") + assert client.get("/assets/app.js").text == "console.log(1)" + assert "home" in client.get("/orders/42/edit").text + assert client.get("/assets/missing.js").status_code == 404 + assert client.get("/api/nothing-here").json() == {"error": "not found", "code": "not_found"} + assert client.get("/../secret.txt").status_code == 404 + assert client.post("/orders").status_code == 404 + + +def test_the_fallback_never_shadows_api_health_or_reserved_paths(client: TestClient) -> None: + assert client.get("/api/orders").json() == [] + assert client.put("/api/orders").status_code == 405 + assert client.get("/api/health").json() == {"status": "ok"} + for path in ("/agentenv", "/agentenv/anything", "/agentenv/ext/website/nope", "/agentenv/../agentenv/x"): + assert client.get(path).status_code in (404, 405), path + assert client.get("/.well-known/agent-env.json").json()["name"] == "Shop" + + +def test_health_waits_for_every_ready_check(dist: Path) -> None: + class Slow(site(dist)): + ready = False + + @ready_check + def database_up(self) -> bool: + return self.ready + + @ready_check + async def cache_warm(self) -> bool: + return True + + env = Slow() + client = TestClient(env.create_app()) + waiting = client.get("/api/health") + assert waiting.status_code == 503 and waiting.json() == {"status": "starting", "waiting_for": ["database_up"]} + env.ready = True + assert client.get("/api/health").status_code == 200 + + +def test_a_declared_health_path_is_served_there(dist: Path) -> None: + client = TestClient(site(dist, health="/healthz")().create_app()) + assert client.get("/healthz").json() == {"status": "ok"} + card = client.get("/.well-known/agent-env.json").json() + assert _extension(card, WEBSITE_EXTENSION_URI)["params"]["health"] == "/healthz" + + +def _extension(card: dict, uri: str) -> dict: + return next(e for e in card["capabilities"]["extensions"] if e["uri"] == uri) + + +def test_the_card_declares_website_v1_its_routes_clock_and_data_plane(client: TestClient) -> None: + card = client.get("/.well-known/agent-env.json").json() + assert card["capabilities"]["operations"] == ["data/reset", "data/add", "data/get"] + params = _extension(card, WEBSITE_EXTENSION_URI)["params"] + assert params["domains"] == ["shop.example", "www.shop.example"] + assert params["health"] == "/api/health" + assert params["routes"] == [ + {"method": "GET", "path": "/api/orders", "name": "list_orders", "roles": None}, + {"method": "POST", "path": "/api/orders", "name": "place_order", "roles": ["default", "agent"]}, + {"method": "POST", "path": "/api/sign-in", "name": "sign_in", "roles": None}, + ] + assert params["methods"]["openapi"] == {"method": "GET", "endpoint": "/agentenv/ext/website/openapi", + "request": {"type": "object", "properties": {}}} + assert params["methods"]["session"]["endpoint"] == "/agentenv/ext/website/session" + clock = _extension(card, "urn:agentenv:clock/v1")["params"] + assert clock["endpoint"] == "/agentenv/ext/sync_time" + + +def test_the_openapi_document_names_each_route(client: TestClient) -> None: + document = client.get("/agentenv/ext/website/openapi").json() + assert document["openapi"].startswith("3.") and document["info"]["title"] == "Shop" + listed = document["paths"]["/api/orders"]["get"] + assert listed["operationId"] == "list_orders" and listed["summary"] == "List the signed-in user's orders." + placed = document["paths"]["/api/orders"]["post"] + assert placed["x-agentenv-roles"] == ["default", "agent"] + assert placed["requestBody"]["content"]["application/json"]["schema"]["type"] == "object" + assert listed["parameters"] == [] + + +def test_the_default_store_round_trips_through_the_data_plane(client: TestClient, shop: WebsiteEnvironment) -> None: + assert rpc(client, "data/get")["parts"][0]["data"] == SEED + rpc(client, "data/reset") + assert rpc(client, "data/get")["parts"][0]["data"] == {} + as_file = base64.b64encode(json.dumps({"orders": [{"id": 7, "user": "carol"}]}).encode()).decode() + rpc(client, "data/add", {"parts": [{"kind": "data", "data": {"current_user": "carol", "users": [{"id": "carol"}]}}, + {"kind": "file", "file": {"bytes": as_file, "name": "orders.json"}}, + {"kind": "text", "text": json.dumps({"notes": [{"id": "n1"}]})}]}) + assert rpc(client, "data/get")["parts"][0]["data"] == { + "current_user": "carol", "users": [{"id": "carol"}], "orders": [{"id": 7, "user": "carol"}], + "notes": [{"id": "n1"}]} + assert shop.acting_user() == "carol" + + +def test_go_reports_initial_current_and_diff_state(client: TestClient) -> None: + client.post("/api/orders", json={"sku": "mug"}) + state = client.get("/go", params={"sid": "default"}).json() + assert state["initial_state"] == SEED + assert state["current_state"]["orders"][-1]["sku"] == "mug" + assert state["state_diff"] == {"orders": {"added": [state["current_state"]["orders"][-1]], "removed": [], + "changed": []}} + + +def test_a_site_with_its_own_hooks_keeps_them_and_serves_no_go(dist: Path) -> None: + class Own(site(dist)): + @get_data + def export(self) -> list: + return [DataPart(data={"mine": True})] + + client = TestClient(Own().create_app()) + assert rpc(client, "data/get")["parts"][0]["data"] == {"mine": True} + assert client.get("/.well-known/agent-env.json").json()["capabilities"]["operations"] == ["data/reset", "data/add", "data/get"] + assert "home" in client.get("/go").text + + +def test_a_seed_data_file_is_the_initial_state(tmp_path: Path, dist: Path) -> None: + (tmp_path / "seed.json").write_text(json.dumps(SEED)) + env = site(dist, data=str(tmp_path / "seed.json"))() + assert env.store.export() == SEED and env.store.initial == SEED and env.acting_user() == "alice" + + +def test_the_acting_user_is_signed_in_until_a_session_says_else(client: TestClient) -> None: + assert client.get("/api/orders").json() == [] + first = client.get("/api/session") + assert first.json() == {"user": "alice", "role": "default", "signed_in": True} + cookie = first.headers["set-cookie"] + assert cookie.startswith(f"{SESSION_COOKIE}=") and "domain" not in cookie.lower() and "httponly" in cookie.lower() + client.post("/api/sign-in", json={"user": "bob"}) + assert client.get("/api/orders").json() == [{"id": 1, "user": "bob"}] + assert client.get("/api/session").json()["user"] == "bob" + assert client.delete("/api/session").json() == {"user": None, "role": "default", "signed_in": False} + assert client.get("/api/session").json()["user"] is None + client.cookies.set(SESSION_COOKIE, "forged.signature") + assert client.get("/api/session").json()["user"] == "alice" + + +def test_the_session_cookie_expires_when_the_session_does(client: TestClient) -> None: + from email.utils import parsedate_to_datetime + + cookie = client.get("/api/session").headers["set-cookie"] + expires = next(a.split("=", 1)[1] for a in cookie.split(";") if a.strip().lower().startswith("expires=")) + ahead = parsedate_to_datetime(expires) - datetime.datetime.now(datetime.timezone.utc) + assert datetime.timedelta(days=7) - datetime.timedelta(minutes=1) < ahead <= datetime.timedelta(days=7) + + +def test_the_session_operation_mints_cookies_for_a_user_or_signed_out(shop: WebsiteEnvironment) -> None: + client = TestClient(shop.create_app()) + minted = client.post("/agentenv/ext/website/session", json={"user": "bob"}).json() + assert minted["user"] == "bob" and minted["expires_at"].endswith("Z") + (cookie,) = minted["cookies"] + assert cookie["name"] == SESSION_COOKIE and "domain" not in cookie and cookie["path"] == "/" + browser = TestClient(shop.asgi, cookies={SESSION_COOKIE: cookie["value"]}) + assert browser.get("/api/session").json()["user"] == "bob" + signed_out = client.post("/agentenv/ext/website/session", json={}).json()["cookies"][0]["value"] + assert TestClient(shop.asgi, cookies={SESSION_COOKIE: signed_out}).get("/api/session").json()["user"] is None + + +def test_every_answer_carries_the_universe_time(client: TestClient) -> None: + for response in (client.get("/"), client.get("/api/orders"), client.get("/nope.js"), + client.post(RPC, json={"jsonrpc": "2.0", "id": 1, "method": "data/get"}), + client.get("/.well-known/agent-env.json")): + now = parse_time(response.headers[UNIVERSE_NOW_HEADER]) + assert abs((now - datetime.datetime.now(datetime.timezone.utc)).total_seconds()) < 60 + + +class _Clock: + """A gateway-shaped /clock/time whose time a test sets.""" + + def __init__(self, virtual: str) -> None: + self.virtual = virtual + outer = self + + class Handler(BaseHTTPRequestHandler): + def do_GET(self) -> None: + body = json.dumps({"virtual_time": outer.virtual}).encode() + self.send_response(200) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, *args) -> None: + return None + + self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + threading.Thread(target=self.server.serve_forever, daemon=True).start() + self.url = f"http://127.0.0.1:{self.server.server_port}/clock/time" + + def close(self) -> None: + self.server.shutdown() + self.server.server_close() + + +@pytest.fixture +def gateway_clock(): + clock = _Clock("2031-01-01T00:00:00Z") + yield clock + clock.close() + + +def test_sync_time_ties_now_and_the_header_to_the_gateway_clock(client: TestClient, shop: WebsiteEnvironment, + gateway_clock: _Clock, monkeypatch) -> None: + synced = client.post("/agentenv/ext/sync_time", json={"env_get_time_url": gateway_clock.url}).json() + assert synced == {"env_get_time_url": gateway_clock.url, "gateway_virtual_time": "2031-01-01T00:00:00.000Z"} + assert shop.now().year == 2031 + assert client.get("/").headers[UNIVERSE_NOW_HEADER].startswith("2031-01-01T00:00:0") + order = client.post("/api/orders", json={"sku": "pen"}).json() + assert order["at"].startswith("2031-01-01") + gateway_clock.virtual = "2031-06-01T12:00:00Z" + monkeypatch.setattr(clock_module, "REFRESH_S", 0.0) + assert client.get("/").headers[UNIVERSE_NOW_HEADER].startswith("2031-06-01T12:00") + + +def test_a_clock_that_cannot_be_read_is_refused_and_a_lost_one_extrapolates(client: TestClient, + shop: WebsiteEnvironment, + gateway_clock: _Clock, monkeypatch) -> None: + refused = client.post("/agentenv/ext/sync_time", json={"env_get_time_url": "http://127.0.0.1:9/clock/time"}) + assert refused.status_code == 500 and shop.now().year == datetime.datetime.now().year + client.post("/agentenv/ext/sync_time", json={"env_get_time_url": gateway_clock.url}) + gateway_clock.close() + monkeypatch.setattr(clock_module, "REFRESH_S", 0.0) + assert client.get("/").headers[UNIVERSE_NOW_HEADER].startswith("2031-01-01") + + +def test_clock_patch_puts_the_clock_script_in_html_only(dist: Path, gateway_clock: _Clock) -> None: + client = TestClient(site(dist, clock_patch=True)().create_app()) + client.post("/agentenv/ext/sync_time", json={"env_get_time_url": gateway_clock.url}) + page = client.get("/") + head = page.text.index("") + len("") + assert page.text[head:].startswith("