feat(website): WebsiteEnvironment serves a site on the website/v1 contract, and a conformance check tests any server (FD-4683)

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 <url>`` 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",
+]

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": <RFC3339>}``,
+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 ("<script>(function(){var D=Date,o=%s-D.now();function n(){return D.now()+o}"
+            "class U extends D{constructor(...a){a.length?super(...a):super(n())}static now(){return n()}}"
+            "window.Date=U;var f=window.fetch;if(f)window.fetch=function(){return f.apply(this,arguments)"
+            ".then(function(r){var t=D.parse(r.headers.get(%s)||'');if(!isNaN(t))o=t-D.now();return r})}})();"
+            "</script>" % (int(now.timestamp() * 1000), json.dumps(UNIVERSE_NOW_HEADER))).encode()

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 <url>`` 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())