A2App passthrough and WebSocket paths

Author: ahmad-ajmal

app/agent_app/a2app_proxy.py

@@ -4,26 +4,38 @@
 AS-IS and cannot host the PocketBase adapter hooks, so the A2App surface
 sits in FRONT of it: the app binds a hidden internal loopback port, this
 proxy binds the project's assigned port, answers the protocol endpoints
-itself, and passes every other request through untouched (the app's own UI
-keeps working). Because the proxy is system code running inside CraftBot,
+itself, and passes every other request through to the app (the app's own
+UI keeps working). Because the proxy is system code running inside CraftBot,
 "adapter stamped at every launch" holds for externals with no sync step.
 
 Served surface (mirrors the native pb_hooks adapter):
   GET /api/_a2app            identity (+ flavor:"external")
   GET /api/_a2app/describe   operations + conventions (entities: {} — the
                              foreign data model is not mapped; ops only)
   GET /api/_ops              operations.json verbatim
-  *   /api/ops/{name}        guarded invocation, mapped onto the app's API
-  *   anything else          transparent passthrough (HTTP + WebSocket)
-
-Auth mirrors _system.pb.js (see `guard_request`). Two independent checks:
-the ORIGIN check refuses foreign-origin mutations outright, and the CALLER
-check requires every mutation to carry a credential — X-A2App-Token from the
-project's .agent-token (programs, the agent), or the UI session cookie the
-app's own browser UI is issued. An allowed Origin is never a credential:
-tunnel traffic arrives over loopback and can claim any Origin it likes.
-Through a tunnel, every request needs the credential, reads included; the
-UI session there is only issued in exchange for the share link's secret.
+  *   /api/ops/{name}        invocation, mapped onto the app's API
+  *   anything else          passthrough to the app (HTTP + WebSocket)
+
+EVERY route, passthrough included, sits behind one guard (`guard_request`,
+mirroring _system.pb.js). Two independent checks: the ORIGIN check refuses
+foreign-origin mutations outright, and the CALLER check requires every
+mutation to carry a credential — X-A2App-Token from the project's
+.agent-token (programs, the agent), or the UI session cookie the app's own
+browser UI is issued with the HTML page that boots it. An allowed Origin is
+never a credential: tunnel traffic arrives over loopback and can claim any
+Origin it likes. Locally, reads are open; through a tunnel, every request
+needs the credential, reads included, and the UI session there is only
+issued in exchange for the share link's secret. A WebSocket handshake is
+judged as a mutation (it opens a write channel, and browsers apply no CORS
+to it) and refused before upgrading. Declared ops are the SHAPED write path
+(typed params, audit log); the app's native API is not a way around the
+guard.
+
+CORS is the proxy's policy, not the app's: upstream Access-Control-* headers
+are stripped, the grant is reflected only to loopback / the shared origin
+(`origin_allowed`), and a foreign origin gets none, so the browser withholds
+every response from it.
+
 Ops are (re)read from operations.json on every request, like the native
 describe, so the surface can never drift from the file on disk.
 """
@@ -52,6 +64,16 @@
 EXTERNAL_ADAPTER_VERSION = "0.1.0"
 LOOPBACK_ORIGIN = re.compile(r"^https?://(127\.0\.0\.1|localhost|\[::1\])(:\d+)?$")
 MUTATING = {"POST", "PUT", "PATCH", "DELETE"}
+# Upstream Access-Control-* kept for ALLOWED origins only (see
+# _passthrough_cors); Allow-Origin itself is always the proxy's.
+UPSTREAM_CORS_SHAPE = {
+    "access-control-allow-methods",
+    "access-control-allow-headers",
+    "access-control-expose-headers",
+    "access-control-max-age",
+    "access-control-allow-credentials",
+    "access-control-allow-private-network",
+}
 # Hop-by-hop headers never forwarded in either direction (RFC 9110 §7.6.1).
 HOP_HEADERS = {
     "connection",
@@ -426,10 +448,15 @@ def _ops_raw(self) -> bytes:
     def _origin_allowed(self, origin: str) -> bool:
         return origin_allowed(self.project_dir, origin)
 
-    def _deny(self, request):
-        """guard_request as a response: None to proceed, else the refusal."""
+    def _deny(self, request, method: Optional[str] = None):
+        """guard_request as a response: None to proceed, else the refusal.
+        `method` overrides the request's (a WebSocket handshake is a GET that
+        opens a write channel, so it is judged as one)."""
         denied = guard_request(
-            self.project_dir, request.method, request.headers, request.cookies
+            self.project_dir,
+            method or request.method,
+            request.headers,
+            request.cookies,
         )
         if denied is None:
             return None
@@ -449,7 +476,24 @@ def _reflect_cors(self, request, resp) -> None:
         origin = request.headers.get("Origin", "")
         if origin and self._origin_allowed(origin):
             resp.headers["Access-Control-Allow-Origin"] = origin
-            resp.headers["Vary"] = "Origin"
+            vary = resp.headers.get("Vary", "")
+            if "origin" not in {v.strip().lower() for v in vary.split(",")}:
+                resp.headers["Vary"] = f"{vary}, Origin" if vary else "Origin"
+
+    def _passthrough_cors(self, request, resp, upstream_headers) -> None:
+        """The proxy's CORS policy over the app's own responses. WHO may read
+        is the proxy's call (_reflect_cors), never the app's — an app sending
+        `Access-Control-Allow-Origin: *` would otherwise hand its data to any
+        site. HOW (methods, headers, max-age, credentials) describes the
+        app's API, so an allowed origin still gets the app's answer to that;
+        a foreign origin gets no Access-Control-* at all."""
+        origin = request.headers.get("Origin", "")
+        if not (origin and self._origin_allowed(origin)):
+            return
+        self._reflect_cors(request, resp)
+        for k, v in upstream_headers.items():
+            if k.lower() in UPSTREAM_CORS_SHAPE:
+                resp.headers[k] = v
 
     def _log_action(self, entry: Dict[str, Any]) -> None:
         try:
@@ -486,8 +530,14 @@ async def _handle(self, request):
             return self._ops_manifest(request)
         if own:
             return await self._invoke(request)
-        # TODO(passthrough-auth): the app's own surface is not guarded yet;
-        # the follow-up applies guard_request here too.
+        # The app's own surface sits under the same guard: declared ops are
+        # not a guarded write path if the native API beside them is open.
+        # A WebSocket handshake is refused BEFORE upgrading — browsers apply
+        # no CORS to it, so this is the only cross-site defence it has.
+        websocket = request.headers.get("Upgrade", "").lower() == "websocket"
+        denied = self._deny(request, method="POST" if websocket else None)
+        if denied is not None:
+            return denied
         return await self._passthrough(request)
 
     def _share_exchange(self, request):
@@ -821,9 +871,12 @@ async def _passthrough(self, request):
                 allow_redirects=False,
             ) as up:
                 resp = web.StreamResponse(status=up.status)
+                # add(), not assignment: an app may send several Set-Cookie.
                 for k, v in up.headers.items():
-                    if k.lower() not in HOP_HEADERS:
-                        resp.headers[k] = v
+                    lk = k.lower()
+                    if lk not in HOP_HEADERS and not lk.startswith("access-control-"):
+                        resp.headers.add(k, v)
+                self._passthrough_cors(request, resp, up.headers)
                 # The app's own UI gets its session with the page that boots
                 # it; its same-origin fetches then carry it automatically.
                 if request.method == "GET" and (up.content_type or "").startswith(
@@ -855,6 +908,8 @@ async def _passthrough(self, request):
             )
 
     async def _ws_passthrough(self, request):
+        # Caller already cleared guard_request (as a mutation) in _handle;
+        # nothing here may upgrade a request that did not pass through it.
         import asyncio
 
         import aiohttp

app/agent_app/test_a2app_external.py

@@ -209,7 +209,31 @@ async def wipe(_request):
         seen["todos"] = []
         return web.json_response({"ok": True})
 
+    # The app's native API as a typical adopted codebase ships it: wide-open
+    # CORS (the thing the proxy must NOT reflect) and more than one cookie.
+    async def echo(request):
+        seen.setdefault("echo", []).append(request.method)
+        resp = web.json_response({"method": request.method})
+        resp.headers["Access-Control-Allow-Origin"] = "*"
+        resp.headers["Access-Control-Allow-Credentials"] = "true"
+        resp.headers["Access-Control-Allow-Methods"] = "GET, POST, PUT, PATCH, DELETE"
+        resp.headers["Access-Control-Allow-Headers"] = "Content-Type, X-Custom"
+        resp.headers["Vary"] = "Accept-Encoding"
+        resp.set_cookie("app_a", "1")
+        resp.set_cookie("app_b", "2")
+        return resp
+
+    async def ws_echo(request):
+        ws = web.WebSocketResponse()
+        await ws.prepare(request)
+        seen.setdefault("ws", []).append(request.headers.get("Origin", ""))
+        async for msg in ws:
+            await ws.send_str("echo:" + msg.data)
+        return ws
+
     app = web.Application()
+    app.router.add_route("*", "/api/echo", echo)
+    app.router.add_get("/ws", ws_echo)
     app.router.add_get("/", root)
     app.router.add_post("/api/todos", create_todo)
     app.router.add_get("/api/todos", list_todos)
@@ -376,6 +400,153 @@ async def post(headers, cookie=None, title="x"):
     assert titles == ["agent", "agent+origin", "legacy", "ui", "visitor", "remote agent"], titles
 
 
+async def _passthrough_matrix(http, base: str, tmp: Path, seen) -> None:
+    """The app's NATIVE surface (everything not /api/_a2app*, /api/_ops,
+    /api/ops/*) under the same guard: it used to be passed through with no
+    Origin or caller check, the app's own CORS reflected verbatim, and its
+    WebSocket open to any site (browsers apply no CORS to the handshake)."""
+    import aiohttp
+
+    echo = f"{base}/api/echo"
+    ws_url = f"ws://127.0.0.1:{PROXY_PORT}/ws"
+    loopback = {"Origin": f"http://127.0.0.1:{PROXY_PORT}"}
+    evil = {"Origin": "https://evil.example"}
+
+    def jar(cookie):
+        return {"Cookie": f"{cookie[0]}={cookie[1]}"}
+
+    async def send(method, headers, url=echo):
+        before = len(seen.get("echo", []))
+        async with http.request(method, url, json={"t": 1}, headers=headers) as r:
+            body = await r.json()
+            reached = len(seen.get("echo", [])) > before
+            return r.status, body, reached
+
+    async def ws(headers):
+        """(status, echoed) — status 101 on upgrade, else the refusal."""
+        try:
+            async with http.ws_connect(ws_url, headers=headers) as w:
+                await w.send_str("hi")
+                msg = await w.receive(timeout=5)
+                return 101, msg.data
+        except aiohttp.WSServerHandshakeError as e:
+            return e.status, None
+
+    async with http.get(f"{base}/") as r:
+        local = _set_cookie(r)
+
+    # ── HTTP writes: foreign origins refused, callers must hold a credential
+    for method in ("POST", "PUT", "PATCH", "DELETE"):
+        status, body, reached = await send(method, evil)
+        assert status == 403 and body["code"] == "forbidden_origin", (method, status)
+        assert not reached, f"foreign-origin {method} reached the app"
+    status, _, reached = await send("POST", {**evil, "X-A2App-Token": TOKEN})
+    assert status == 403 and not reached, "a token does not launder a foreign origin"
+    status, _, reached = await send("POST", {**evil, **jar(local)})
+    assert status == 403 and not reached, "a session does not launder a foreign origin"
+    for label, headers in (
+        ("no Origin, no credential", {}),
+        ("loopback Origin, no credential", loopback),
+        ("wrong token", {"X-A2App-Token": "nope"}),
+    ):
+        status, body, reached = await send("POST", headers)
+        assert status == 401 and body["code"] == "unauthorized", (label, status)
+        assert not reached, label
+    async with http.post(f"{base}/api/todos", json={"title": "evil"}, headers=evil) as r:
+        assert r.status == 403, "a real app route, not just the echo"
+
+    # the app's own UI (loopback page + the session its HTML page issued)
+    for method in ("POST", "PUT", "PATCH", "DELETE"):
+        status, body, reached = await send(method, {**loopback, **jar(local)})
+        assert status == 200 and body["method"] == method and reached, method
+    # the agent / a local program
+    status, _, reached = await send("POST", {"X-A2App-Token": TOKEN})
+    assert status == 200 and reached
+    # local reads stay open (CORS, below, keeps them from foreign pages)
+    status, _, reached = await send("GET", {})
+    assert status == 200 and reached
+
+    # ── CORS: the proxy's policy, never the app's `*` ──
+    async with http.get(echo, headers=evil) as r:
+        acx = [k for k in r.headers if k.lower().startswith("access-control-")]
+        assert r.status == 200 and not acx, f"upstream CORS reflected: {acx}"
+    other_local = {"Origin": "http://localhost:5173"}
+    async with http.get(echo, headers=other_local) as r:
+        assert r.headers["Access-Control-Allow-Origin"] == "http://localhost:5173"
+        assert r.headers["Access-Control-Allow-Credentials"] == "true"
+        vary = {v.strip() for v in r.headers["Vary"].split(",")}
+        assert vary == {"Accept-Encoding", "Origin"}, vary
+        cookies = {c.split("=", 1)[0] for c in r.headers.getall("Set-Cookie")}
+        assert {"app_a", "app_b"} <= cookies, "every upstream Set-Cookie survives"
+    preflight = {"Access-Control-Request-Method": "PUT"}
+    async with http.options(echo, headers={**evil, **preflight}) as r:
+        acx = [k for k in r.headers if k.lower().startswith("access-control-")]
+        assert not acx, f"foreign preflight granted: {acx}"
+    async with http.options(echo, headers={**other_local, **preflight}) as r:
+        assert r.headers["Access-Control-Allow-Origin"] == "http://localhost:5173"
+        assert "PUT" in r.headers["Access-Control-Allow-Methods"]
+        assert r.headers["Access-Control-Allow-Headers"] == "Content-Type, X-Custom"
+
+    # ── WebSocket: judged at the handshake, refused before upgrading ──
+    ws_seen = len(seen.get("ws", []))
+    for label, headers, want in (
+        ("foreign Origin", evil, 403),
+        ("foreign Origin + token", {**evil, "X-A2App-Token": TOKEN}, 403),
+        ("foreign Origin + session", {**evil, **jar(local)}, 403),
+        ("no Origin, no credential", {}, 401),
+        ("loopback Origin, no credential", loopback, 401),
+    ):
+        status, _ = await ws(headers)
+        assert status == want, (label, status)
+    assert len(seen.get("ws", [])) == ws_seen, "a refused handshake reached the app"
+    assert await ws({**loopback, **jar(local)}) == (101, "echo:hi"), "the app's own UI"
+    assert await ws({"X-A2App-Token": TOKEN}) == (101, "echo:hi"), "the agent"
+
+    # ── through the tunnel ──
+    secret = "share-secret-for-passthrough-0123456789"
+    (tmp / ".tunnel-origin").write_text(SHARED, encoding="utf-8")
+    (tmp / ".tunnel-secret").write_text(secret, encoding="utf-8")
+    try:
+        async with http.get(f"{base}/", headers=VIA_TUNNEL) as r:
+            body = await r.json()
+            assert r.status == 401 and body["code"] == "share_session_required", (
+                "a bare tunnel URL must not read the app"
+            )
+        async with http.get(
+            f"{base}/?a2app_share={secret}", headers=VIA_TUNNEL, allow_redirects=False
+        ) as r:
+            assert r.status == 302
+            shared = _set_cookie(r)
+        visitor = {**VIA_TUNNEL, "Origin": SHARED, **jar(shared)}
+
+        async with http.get(f"{base}/", headers={**VIA_TUNNEL, **jar(shared)}) as r:
+            assert r.status == 200 and (await r.text()) == "UPSTREAM OK"
+        status, _, reached = await send("POST", visitor)
+        assert status == 200 and reached, "the shared UI writes"
+        async with http.get(echo, headers=visitor) as r:
+            assert r.headers["Access-Control-Allow-Origin"] == SHARED
+        assert await ws(visitor) == (101, "echo:hi"), "the shared UI's WebSocket"
+
+        for label, headers, want in (
+            ("no session", {**VIA_TUNNEL, "Origin": SHARED}, 401),
+            ("forged loopback Origin", {**VIA_TUNNEL, **loopback}, 401),
+            ("local session", {**VIA_TUNNEL, "Origin": SHARED, **jar(local)}, 401),
+            ("foreign Origin + session", {**VIA_TUNNEL, **evil, **jar(shared)}, 403),
+        ):
+            status, _, reached = await send("POST", headers)
+            assert status == want and not reached, ("http", label, status)
+            status, _ = await ws(headers)
+            assert status == want, ("ws", label, status)
+        status, _, reached = await send("POST", {**VIA_TUNNEL, "X-A2App-Token": TOKEN})
+        assert status == 200 and reached, "a remote agent with the token"
+    finally:
+        (tmp / ".tunnel-secret").unlink(missing_ok=True)
+        (tmp / ".tunnel-origin").unlink(missing_ok=True)
+    # Let the proxy finish closing its upstream WebSockets while this loop
+    # (which also runs the upstream) is still free to answer them.
+    await asyncio.sleep(0.3)
+
+
 async def _proxy_suite(tmp: Path) -> None:
     import aiohttp
 
@@ -449,6 +620,7 @@ async def _proxy_suite(tmp: Path) -> None:
             assert r.status == 200 and (await r.json())["title"] == "call John"
 
         await _auth_matrix(http, base, tmp, seen)
+        await _passthrough_matrix(http, base, tmp, seen)
 
         # unknown op -> 404 envelope, never a silent passthrough
         async with http.post(f"{base}/api/ops/nope", json={}, headers=auth) as r: