From 94478113f5daba8d1a8f3f1807508038dfd98beb Mon Sep 17 00:00:00 2001 From: Andrew Miller Date: Sat, 7 Nov 2026 08:52:45 -0500 Subject: [PATCH] Owner-only /_api/logs for any container; ship-fix keeps the pod's logs private GET /_api/logs lists every container on the CVM and GET /_api/logs/ streams one container's full stdout/stderr (tail=all by default, since/until in unix seconds). Both take the owner token only. This replaces Phala's public_logs endpoint, which serves the same logs to anyone. ship-fix.sh passes --no-public-logs for the pod (phala deploy defaults it to true), and refuses to deploy when the local compose differs from the live one in anything but the daemon image. --- proxy/docker_client.py | 34 +++++++++++ proxy/ingress.py | 46 ++++++++++++++- proxy/test_owner_logs.py | 124 +++++++++++++++++++++++++++++++++++++++ ship-fix.sh | 31 +++++++++- 4 files changed, 232 insertions(+), 3 deletions(-) create mode 100644 proxy/test_owner_logs.py diff --git a/proxy/docker_client.py b/proxy/docker_client.py index 6bb43c4f..ba2be3dc 100644 --- a/proxy/docker_client.py +++ b/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: diff --git a/proxy/ingress.py b/proxy/ingress.py index 360d9fe4..7d1bd726 100644 --- a/proxy/ingress.py +++ b/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/: 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//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. diff --git a/proxy/test_owner_logs.py b/proxy/test_owner_logs.py new file mode 100644 index 00000000..9173ecdf --- /dev/null +++ b/proxy/test_owner_logs.py @@ -0,0 +1,124 @@ +"""GET /_api/logs and /_api/logs/: 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") diff --git a/ship-fix.sh b/ship-fix.sh index 83dd04a1..20708f80 100755 --- a/ship-fix.sh +++ b/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/ 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)."