Owner-only /_api/logs for any container; ship-fix keeps the pod's logs private

proxy/docker_client.py

@@ -1,5 +1,6 @@
 """Thin async Docker Engine client over Unix socket."""
 
+import asyncio
 import base64
 import json
 import logging
@@ -263,6 +264,39 @@ async def list_containers(self, label: str) -> list[dict]:
             raise RuntimeError(f"list containers failed ({status}): {data}")
         return data
 
+    async def list_all_containers(self) -> list[dict]:
+        status, data = await self._json_request("GET", "/containers/json", params={"all": "true"})
+        if status >= 400:
+            raise RuntimeError(f"list containers failed ({status}): {data}")
+        return data
+
+    async def stream_logs(self, name: str, params: dict):
+        """Yield a container's log bytes, demuxed, without holding the history in memory
+        (the pod's daemon log was 860 MB on 2026-11-07). Raises LookupError if there is no
+        such container; nothing has been yielded at that point."""
+        status, info = await self._json_request("GET", f"/containers/{name}/json")
+        if status == 404:
+            raise LookupError(f"no such container: {name}")
+        if status >= 400:
+            raise RuntimeError(f"inspect failed ({status}): {info}")
+        conn = aiohttp.UnixConnector(path=self.socket_path)
+        async with aiohttp.ClientSession(connector=conn, timeout=aiohttp.ClientTimeout(total=None)) as session:
+            async with session.get(f"http://localhost/containers/{name}/logs", params=params) as resp:
+                if resp.status >= 400:
+                    raise RuntimeError(f"logs failed ({resp.status}): {await resp.text()}")
+                if info["Config"]["Tty"]:
+                    async for chunk in resp.content.iter_chunked(1 << 16):
+                        yield chunk
+                    return
+                while True:
+                    try:
+                        header = await resp.content.readexactly(8)
+                    except asyncio.IncompleteReadError as e:
+                        if e.partial:
+                            raise
+                        return
+                    yield await resp.content.readexactly(int.from_bytes(header[4:8], "big"))
+
     async def stats(self, cid: str) -> dict:
         status, data = await self._json_request("GET", f"/containers/{cid}/stats?stream=false", timeout=30)
         if status >= 400:

proxy/ingress.py

@@ -8,6 +8,7 @@
 import json
 import logging
 import os
+import re
 import time
 from dataclasses import asdict
 from datetime import datetime, timezone
@@ -62,6 +63,12 @@ def _redact_env(data: dict) -> dict:
 DSTACK_SOCK = None  # set by main.py
 API_TOKEN = os.environ.get("TEE_DAEMON_TOKEN", "")
 
+# /_api/logs/<name>: the name goes into a Docker API path, so only Docker's own name alphabet
+# (dstack guest-agent had to add the same check, Dstack-TEE/dstack#1443).
+_CONTAINER_NAME_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,127}")
+_LOG_TAIL_RE = re.compile(r"all|[0-9]{1,9}")
+_UNIX_TS_RE = re.compile(r"[0-9]{1,10}(\.[0-9]{1,9})?")
+
 # Headers this proxy never forwards upstream.
 HOP_HEADERS = ("host", "transfer-encoding", "accept-encoding")
 
@@ -679,7 +686,8 @@ async def _handle_api(self, request: web.Request, path: str) -> web.Response:
                     resp.headers["Access-Control-Allow-Origin"] = "*"
                     return resp
 
-        owner_only = path in ("tokens", "stats") or path.startswith("tokens/")
+        owner_only = (path in ("tokens", "stats", "logs") or path.startswith("tokens/")
+                      or path.startswith("logs/"))
         denied = self._check_auth(request, path, owner_only=owner_only)
         if denied:
             return denied
@@ -723,6 +731,13 @@ async def _handle_api(self, request: web.Request, path: str) -> web.Response:
         if path == "stats" and method == "GET":
             return await self._api_stats()
 
+        if path == "logs" and method == "GET":
+            return web.json_response([
+                {"name": c["Names"][0].lstrip("/"), "state": c["State"], "status": c["Status"],
+                 "image": c["Image"]} for c in await self.docker.list_all_containers()])
+        if path.startswith("logs/") and method == "GET":
+            return await self._api_container_logs(path[len("logs/"):], request)
+
         # RFC 0017: fleet export/import (authed)
         if path == "export" and method == "GET":
             return await self._api_export()
@@ -984,6 +999,35 @@ async def _api_logs(self, name: str, request: web.Request) -> web.Response:
         text = await self.docker.logs(cid, tail=tail)
         return web.Response(text=text, content_type="text/plain")
 
+    async def _api_container_logs(self, name: str, request: web.Request) -> web.StreamResponse:
+        """Owner-only: the full stdout/stderr of ANY container on the CVM, streamed.
+
+        This is the private replacement for Phala's public_logs endpoint. With public_logs on,
+        anyone can download every container's history (the daemon's access log included); with
+        it off, nobody can, the owner included. The daemon already holds the Docker socket and
+        the owner token, so it serves the logs itself. Unlike /_api/projects/<name>/logs this
+        reaches the daemon, the shared runtimes and the compose siblings (ingress), and it is
+        not capped: tail defaults to all, and since/until (unix seconds) bound the window."""
+        q = request.query
+        params = {"stdout": "true", "stderr": "true", "timestamps": "true",
+                  "tail": q.get("tail", "all")}
+        params.update({k: q[k] for k in ("since", "until") if k in q})
+        if (not _CONTAINER_NAME_RE.fullmatch(name) or not _LOG_TAIL_RE.fullmatch(params["tail"])
+                or any(not _UNIX_TS_RE.fullmatch(params[k]) for k in ("since", "until") if k in params)):
+            return web.json_response({"error": "bad container name, tail, since or until"}, status=400)
+        chunks = self.docker.stream_logs(name, params)
+        try:
+            first = await anext(chunks, b"")
+        except LookupError as e:
+            return web.json_response({"error": str(e)}, status=404)
+        resp = web.StreamResponse(headers={"Content-Type": "text/plain; charset=utf-8"})
+        await resp.prepare(request)
+        await resp.write(first)
+        async for chunk in chunks:
+            await resp.write(chunk)
+        await resp.write_eof()
+        return resp
+
     async def _api_stats(self) -> web.Response:
         """Memory/CPU per managed container (issue #151), owner-only, read-only.
 

proxy/test_owner_logs.py

@@ -0,0 +1,124 @@
+"""GET /_api/logs and /_api/logs/<container>: owner-only, streamed container logs."""
+
+import asyncio
+import tempfile
+
+from aiohttp import web
+from aiohttp.test_utils import TestClient, TestServer
+
+from . import ingress
+from .docker_client import DockerClient
+from .ingress import Ingress
+
+
+def _frame(stream, data):
+    return bytes([stream, 0, 0, 0]) + len(data).to_bytes(4, "big") + data
+
+
+class FakeDocker:
+    def __init__(self):
+        self.calls = []
+
+    async def list_all_containers(self):
+        return [{"Names": ["/dstack-tee-daemon-1"], "State": "running", "Status": "Up 7 days",
+                 "Image": "ghcr.io/amiller/tee-socket-proxy@sha256:ab"}]
+
+    async def stream_logs(self, name, params):
+        self.calls.append((name, params))
+        if name != "dstack-tee-daemon-1":
+            raise LookupError(f"no such container: {name}")
+        yield b"line 1\n"
+        yield b"line 2\n"
+
+
+class FakeTokens:
+    def authenticate(self, token, path):
+        return True  # a scoped token that would pass any scope check
+
+
+async def _get(docker, url, token):
+    ing = Ingress(None, docker, None, None, None, None, FakeTokens())
+    app = web.Application()
+    app.router.add_route("*", "/{path:.*}", ing.handle)
+    async with TestClient(TestServer(app)) as client:
+        resp = await client.get(url, headers={"Authorization": f"Bearer {token}"})
+        return resp.status, await resp.text()
+
+
+def test_owner_reads_full_history(monkeypatch):
+    monkeypatch.setattr(ingress, "API_TOKEN", "owner")
+    docker = FakeDocker()
+    status, body = asyncio.run(_get(docker, "/_api/logs/dstack-tee-daemon-1?since=1730419200", "owner"))
+    assert (status, body) == (200, "line 1\nline 2\n")
+    assert docker.calls == [("dstack-tee-daemon-1", {"stdout": "true", "stderr": "true",
+                             "timestamps": "true", "tail": "all", "since": "1730419200"})]
+
+
+def test_list(monkeypatch):
+    monkeypatch.setattr(ingress, "API_TOKEN", "owner")
+    status, body = asyncio.run(_get(FakeDocker(), "/_api/logs", "owner"))
+    assert status == 200 and '"name": "dstack-tee-daemon-1"' in body
+
+
+def test_scoped_token_refused(monkeypatch):
+    monkeypatch.setattr(ingress, "API_TOKEN", "owner")
+    for url in ("/_api/logs", "/_api/logs/dstack-tee-daemon-1"):
+        assert asyncio.run(_get(FakeDocker(), url, "scoped"))[0] == 403
+
+
+def test_bad_input_and_missing(monkeypatch):
+    monkeypatch.setattr(ingress, "API_TOKEN", "owner")
+    for url in ("/_api/logs/..%2Fimages%2Fjson", "/_api/logs/a?tail=-1",
+                "/_api/logs/a?since=yesterday", "/_api/logs/a?until=1%26follow%3Dtrue"):
+        assert asyncio.run(_get(FakeDocker(), url, "owner"))[0] == 400, url
+    assert asyncio.run(_get(FakeDocker(), "/_api/logs/nope", "owner"))[0] == 404
+
+
+async def _engine(sock, tty, body):
+    async def inspect(req):
+        if req.match_info["name"] != "c":
+            return web.json_response({"message": "No such container"}, status=404)
+        return web.json_response({"Config": {"Tty": tty}})
+
+    async def logs(req):
+        resp = web.StreamResponse()
+        await resp.prepare(req)
+        for i in range(0, len(body), 5):  # split frames across writes
+            await resp.write(body[i:i + 5])
+        await resp.write_eof()
+        return resp
+
+    app = web.Application()
+    app.router.add_get("/containers/{name}/json", inspect)
+    app.router.add_get("/containers/{name}/logs", logs)
+    runner = web.AppRunner(app)
+    await runner.setup()
+    await web.UnixSite(runner, sock).start()
+    return runner
+
+
+async def _collect(tty, body, name="c"):
+    with tempfile.TemporaryDirectory() as d:
+        sock = f"{d}/docker.sock"
+        runner = await _engine(sock, tty, body)
+        try:
+            return b"".join([c async for c in DockerClient(sock).stream_logs(name, {"tail": "all"})])
+        finally:
+            await runner.cleanup()
+
+
+def test_stream_logs_demuxes_real_framing():
+    framed = _frame(1, b"out a\n") + _frame(2, b"err b\n") + _frame(1, b"x" * 70000)
+    assert asyncio.run(_collect(False, framed)) == b"out a\nerr b\n" + b"x" * 70000
+
+
+def test_stream_logs_tty_is_raw():
+    assert asyncio.run(_collect(True, b"raw tty bytes\n")) == b"raw tty bytes\n"
+
+
+def test_stream_logs_missing_container():
+    try:
+        asyncio.run(_collect(False, b"", name="nope"))
+    except LookupError:
+        return
+    raise AssertionError("expected LookupError")

ship-fix.sh

@@ -13,6 +13,10 @@ HA="$HOME/projects/hermes-agent"         # where the deploy manifests live
 IMG="ghcr.io/amiller/tee-socket-proxy"
 KMS=()
 PRELAUNCH=()
+# `phala deploy` defaults --public-logs to TRUE, so every deploy without the flag re-publishes
+# every container's full log (the daemon's access log included) to anyone. The pod is private;
+# the owner reads logs through the daemon's /_api/logs/<container> instead.
+LOGS=()
 
 case "$TARGET" in
   staging) CVM=webhost-staging; COMPOSE="$HA/docker-compose.webhost-staging.yaml"; ENVF="$HA/deploy-notes/.env.webhost-staging" ;;
@@ -25,7 +29,8 @@ case "$TARGET" in
            # boot has no such runtime, so verify_configured_runtime() refuses to start the daemon
            # and the pod is dark. It also carries the wider default-address-pools.
            PRELAUNCH=(--pre-launch-script "$HA/deploy-notes/prelaunch-runsc-resilient.sh")
-           [ -f "$HA/deploy-notes/prelaunch-runsc-resilient.sh" ] || { echo "missing prelaunch script — refusing to deploy a pod that would boot without runsc" >&2; exit 1; } ;;
+           [ -f "$HA/deploy-notes/prelaunch-runsc-resilient.sh" ] || { echo "missing prelaunch script — refusing to deploy a pod that would boot without runsc" >&2; exit 1; }
+           LOGS=(--no-public-logs) ;;
   *) echo "unknown target '$TARGET' (valid: staging, prod, pod)" >&2; exit 1 ;;
 esac
 COMMIT=$(git -C "$TEE" rev-parse --short HEAD)
@@ -34,6 +39,28 @@ TAG="$TARGET-$COMMIT"
 [ -f "$COMPOSE" ] || { echo "missing compose: $COMPOSE" >&2; exit 1; }
 [ -f "$ENVF" ]    || { echo "missing env (not present in this environment?): $ENVF" >&2; exit 1; }
 
+echo "==> 0/4 local compose must match the live one except for the daemon image"
+# This script only swaps the daemon digest, so anything else in $COMPOSE that differs from what
+# the CVM runs gets deployed silently. On 2026-11-07 the local prod7 compose was months behind the
+# pod (older daemon, no PROJECT_HOSTS, one ingress domain instead of two): shipping it would have
+# broken private-agent's domains. Fix the local file from the live one, then rerun.
+phala cvms get "$CVM" --json | python3 -c '
+import json, sys, yaml
+def norm(text):
+    c = yaml.safe_load(text)
+    c["services"]["tee-daemon"].pop("image")
+    return c
+live = norm(json.load(sys.stdin)["compose_file"]["docker_compose_file"])
+local = norm(open(sys.argv[1]).read())
+if live != local:
+    for k in sorted(set(live["services"]) | set(local["services"])):
+        if live["services"].get(k) != local["services"].get(k):
+            print(f"  service {k!r} differs", file=sys.stderr)
+    if live.get("volumes") != local.get("volumes"):
+        print("  volumes differ", file=sys.stderr)
+    sys.exit(f"{sys.argv[1]} drifted from the live {sys.argv[2]} compose — refusing to deploy")
+print("    matches live")' "$COMPOSE" "$CVM"
+
 echo "==> 1/4 build patched image from canonical source ($COMMIT)"
 docker build --build-arg GIT_COMMIT="$COMMIT" -t "$IMG:$TAG" "$TEE"
 
@@ -55,7 +82,7 @@ sed -i -E "s#${IMG}@sha256:[a-f0-9]+#${DIGEST}#" "$COMPOSE"
 grep -n "$IMG" "$COMPOSE"
 
 echo "==> 4/4 upgrade $CVM CVM (restarts every service in the compose; sealed env re-supplied)"
-phala deploy --cvm-id "$CVM" -c "$COMPOSE" -e "$ENVF" "${KMS[@]}" "${PRELAUNCH[@]}" --wait
+phala deploy --cvm-id "$CVM" -c "$COMPOSE" -e "$ENVF" "${KMS[@]}" "${PRELAUNCH[@]}" "${LOGS[@]}" --wait
 
 echo
 echo "DONE ($TARGET -> $CVM)."