diff --git a/.github/workflows/build-os-images.yml b/.github/workflows/build-os-images.yml index db7a691fe..2cd58b984 100644 --- a/.github/workflows/build-os-images.yml +++ b/.github/workflows/build-os-images.yml @@ -27,6 +27,10 @@ on: description: sha-pinned silkd carrier ref (empty = unchanged) type: string default: "" + base: + description: commit the last green images run built (empty = build all) + type: string + default: "" no_cache: type: boolean default: false @@ -63,7 +67,7 @@ jobs: CARRIER_CHANGED: ${{ inputs.boot_changed || inputs.silkd_changed }} MANUAL: ${{ github.event_name == 'workflow_dispatch' }} MANUAL_NO_CACHE: ${{ github.event.inputs.no-cache || inputs.no_cache }} - BEFORE: ${{ github.event.before }} + BASE: ${{ inputs.base }} SHA: ${{ github.sha }} run: | all_names() { @@ -76,16 +80,16 @@ jobs: } # A changed carrier (boot/silkd) refolds into every image; a manual - # run, a first push (no BEFORE), or a change to how the images are + # run, no known base, or a change to how the images are # built rebuilds all. Otherwise only the git-changed image dirs # rebuild. The workflow diff is last so a manual run never runs git. wf=.github/workflows - if [[ "$CARRIER_CHANGED" == "true" || "$MANUAL" == "true" || "$BEFORE" == "0000000000000000000000000000000000000000" ]] || - git diff --name-only "$BEFORE" "$SHA" -- "$wf/build-os-images.yml" "$wf/images.yml" | grep -q .; then + if [[ "$CARRIER_CHANGED" == "true" || "$MANUAL" == "true" || -z "$BASE" ]] || + git diff --name-only "$BASE" "$SHA" -- "$wf/build-os-images.yml" "$wf/images.yml" | grep -q .; then all_names all_changed="" else - all_changed=$(git diff --name-only "$BEFORE" "$SHA" -- 'os-image/' ':!*.md' | sort -u) + all_changed=$(git diff --name-only "$BASE" "$SHA" -- 'os-image/' ':!*.md' | sort -u) declare -A names for f in $all_changed; do name=$(echo "$f" | cut -d'/' -f2) @@ -319,8 +323,8 @@ jobs: id: epoch run: echo "epoch=$(git log -1 --format=%ct)" >> "$GITHUB_OUTPUT" - # SOURCE_DATE_EPOCH plus rewrite-timestamp clamp every file's mtime, so a - # rebuild of one commit writes the same layer. + # SOURCE_DATE_EPOCH plus rewrite-timestamp clamp every file's mtime; a rebuild writes + # the same layer only when the packages it installs (apt, unpinned pip) are unchanged. - name: Build and push uses: docker/build-push-action@v7.4.0 env: diff --git a/.github/workflows/images.yml b/.github/workflows/images.yml index cb465df7b..440a9aeee 100644 --- a/.github/workflows/images.yml +++ b/.github/workflows/images.yml @@ -32,6 +32,7 @@ concurrency: cancel-in-progress: false permissions: + actions: read contents: read packages: write @@ -42,6 +43,7 @@ jobs: boot: ${{ steps.diff.outputs.boot }} silkd: ${{ steps.diff.outputs.silkd }} osimage: ${{ steps.diff.outputs.osimage }} + base: ${{ steps.diff.outputs.base }} steps: - name: Checkout uses: actions/checkout@v7.0.1 @@ -52,14 +54,20 @@ jobs: id: diff env: EVT: ${{ github.event_name }} - BEFORE: ${{ github.event.before }} SHA: ${{ github.sha }} + GH_TOKEN: ${{ github.token }} + GH_REPO: ${{ github.repository }} + BRANCH: ${{ github.ref_name }} run: | - # Manual run or a first/force push with no base: build everything. - if [[ "$EVT" == "workflow_dispatch" || -z "$BEFORE" || "$BEFORE" == "0000000000000000000000000000000000000000" ]]; then - boot=true; silkd=true; osimage=true + # The base is the last green run's commit, not event.before: the queue cancels a pending push unbuilt. + base="" + if [[ "$EVT" != "workflow_dispatch" ]]; then + base=$(gh run list --workflow images.yml --branch "$BRANCH" --status success --limit 1 --json headSha --jq '.[0].headSha // ""') + fi + if [[ -z "$base" ]] || ! git cat-file -e "$base^{commit}" 2>/dev/null; then + base=""; boot=true; silkd=true; osimage=true else - changed() { git diff --name-only "$BEFORE" "$SHA" -- "$@" ':!*.md' | grep -q .; } + changed() { git diff --name-only "$base" "$SHA" -- "$@" ':!*.md' | grep -q .; } # Each area carries its own build workflow: changing how an image is # built must rebuild it, or the run goes green having built nothing. wf=.github/workflows @@ -71,8 +79,9 @@ jobs: echo "boot=$boot" echo "silkd=$silkd" echo "osimage=$osimage" + echo "base=$base" } >> "$GITHUB_OUTPUT" - echo "Detected: boot=$boot silkd=$silkd osimage=$osimage" + echo "Detected: base=${base:-none} boot=$boot silkd=$silkd osimage=$osimage" boot: needs: detect @@ -102,5 +111,6 @@ jobs: silkd_changed: ${{ needs.detect.outputs.silkd == 'true' }} boot_image: ${{ needs.boot.outputs.image }} silkd_image: ${{ needs.silkd.outputs.image }} + base: ${{ needs.detect.outputs.base }} no_cache: ${{ github.event.inputs.no-cache == 'true' }} secrets: inherit diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index d84bbe3cb..7687453ef 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -17,6 +17,10 @@ jobs: uses: actions/checkout@v7.0.1 with: fetch-depth: 0 + - name: Check tag matches silkd version + run: | + crate_ver=$(sed -n 's/^version = "\(.*\)"/\1/p' silkd/Cargo.toml | head -1) + [ "$crate_ver" = "${GITHUB_REF_NAME#v}" ] || { echo "tag ${GITHUB_REF_NAME#v} != silkd/Cargo.toml $crate_ver" >&2; exit 1; } - name: Set up Go uses: actions/setup-go@v7.0.0 with: diff --git a/Makefile b/Makefile index c4dabce94..fe6ba27af 100644 --- a/Makefile +++ b/Makefile @@ -60,7 +60,7 @@ sandboxd: ## build dist/sandboxd go-test: ## go test -race across the Go modules for m in $(GO_MODULES); do (cd $$m && GOWORK=off go test -race ./...) || exit 1; done -bench: ## claim-tier + data-plane benchmarks on this node (see docs/benchmarks.md) +bench: ## claim-tier + data-plane benchmarks on this node; removes every sbx-* VM (see docs/benchmarks.md) bash scripts/bench.sh go-lint: golangci-lint ## golangci-lint (run + fmt --diff) across the Go modules, GOOS linux+darwin diff --git a/docs/benchmarks.md b/docs/benchmarks.md index 66dec788b..4ab46c329 100644 --- a/docs/benchmarks.md +++ b/docs/benchmarks.md @@ -70,6 +70,8 @@ unpooled template, then measures the data plane (`fs_stat` round-trips via in its table so new runs remain comparable with the dated log below. It prints a markdown table stamped with the host evidence: virtualization (`systemd-detect-virt`), CPU, kernel, cocoon version, image digest. +On exit it removes every `sbx-*` VM on the cocoon root, a serving +sandboxd's claims included, so run it only on a node that serves no claims. Knobs (environment variables): `WARM`/`WARM_N` (warm-pool depth and burst size — the burst must stay within the depth, or refill loses the race and diff --git a/docs/deploy.md b/docs/deploy.md index 1e2450078..30af7797e 100644 --- a/docs/deploy.md +++ b/docs/deploy.md @@ -634,6 +634,14 @@ cold boot + snapshot export, tens of seconds) and keeps each pool topped up with claim-ready clones. `GET /v1/info` shows `"golden": true` and `warm` at target when the node is ready to serve warm claims. +A golden left on disk is adopted on start when its stamp matches the pool's +image, lane, interception, `storage` and `warmup`. The stamp does not record +the cocoon or Cloud Hypervisor version, so a golden from before an upgrade of +either is adopted. If its clones then fail, sandboxd retires it once the pool's +refills have failed 8 times running (a node-capacity failure does not count) +and builds a new golden; until it is ready, claims for that pool that find no +warm VM cold-boot. If failures continue with the new golden, it is retired again at most once every +30 s. A minimal systemd unit (shipped as [`packaging/sandboxd.service`](https://github.com/cocoonstack/sandbox/blob/main/packaging/sandboxd.service)): @@ -647,6 +655,7 @@ After=network-online.target [Service] ExecStart=/usr/local/bin/sandboxd -config /etc/sandboxd/config.json Restart=on-failure +RestartSec=5s KillMode=process Environment=SANDBOXD_LOG_LEVEL=info diff --git a/docs/sandboxd-api.md b/docs/sandboxd-api.md index ec4055369..14ac12150 100644 --- a/docs/sandboxd-api.md +++ b/docs/sandboxd-api.md @@ -409,7 +409,9 @@ timeouts return 504 and cancellation returns 408. A mounted writable attachment marks the image dirty before the guest can write. Live removal and release clear the marker only when the unmount observes and normally unmounts the exact writable whole-device filesystem and finds no -residual device mounts. Live removal also needs a confirmed host detach. Both +residual device mounts. These checks read silkd's own mount namespace, so a +mount of the device in another namespace, or one moved there, is not seen. +Live removal also needs a confirmed host detach. Both mount and superblock options must be writable; ext4's `emergency_ro` after a filesystem error counts as read-only. A missing device or pre-unmounted filesystem does not prove clean unmount. The marker remains on uncertainty and @@ -1146,7 +1148,9 @@ connection to its silkd with the unread bytes replayed, so a client sees one silkd and no close. When the sandbox sleeps again meanwhile (a new hibernate or an archive), the connection's next read moves to a fresh reader of the new sleep; when it wakes, the connection moves to the woken sandbox, so reads never come -from a stale sleep. Files +from a stale sleep. Pipelining a request behind a read that has not yet sent +its terminal frame is unsupported: when that request moves the connection, the +reader closes and the rest of the read's reply is lost. Files and symlinks resolve inside the sandbox's own filesystem as of the guest `sync` that runs right before the last hibernate's pause; a write the guest made in the instant between that `sync` and the pause is visible only after a wake. A diff --git a/docs/silkd.md b/docs/silkd.md index 1ea79925c..2c9016819 100644 --- a/docs/silkd.md +++ b/docs/silkd.md @@ -23,7 +23,9 @@ frames tagged `"type"` until a terminal one (`exit`, `done`, or `error`), then reads the next request on the same connection. `stdin`, `stdin_close`, `data` and `data_end` frames are input for the RPC in flight: after its terminal frame they are dropped, and a request frame sent while a streaming -RPC still runs ends that RPC's input and is served after it. A client that +RPC still runs ends that RPC's input and is served after it, except in the +upload of `fs_write` or `fs_push`: there a request frame before `data_end` +fails the upload with `bad_request` and is dropped. A client that closes after one RPC is served exactly as before; `info` reports `proto: 2` on a daemon that keeps the connection. Binary payloads ride base64 in `data` fields. Frames are capped at 8 MiB; requests carry `"v": 1` and unknown diff --git a/packaging/sandboxd.service b/packaging/sandboxd.service index d4f8b5c5c..8318e67b7 100644 --- a/packaging/sandboxd.service +++ b/packaging/sandboxd.service @@ -6,6 +6,7 @@ After=network-online.target [Service] ExecStart=/usr/local/bin/sandboxd -config /etc/sandboxd/config.json Restart=on-failure +RestartSec=5s KillMode=process Environment=SANDBOXD_LOG_LEVEL=info diff --git a/scripts/archive-e2e.sh b/scripts/archive-e2e.sh index 76ecbe11e..bf130b143 100644 --- a/scripts/archive-e2e.sh +++ b/scripts/archive-e2e.sh @@ -2,8 +2,8 @@ # Bare-metal acceptance for lifecycle auto-archive: a claim idles into # hibernation, then to a store checkpoint (its local VM dropped), and the next # access wakes it from the store with the same id and guest state intact. -# Runs a dedicated sandboxd on port 7778 so it never collides with -# sandboxd-e2e.sh (7777). +# Runs a dedicated sandboxd on port 7778 (7777=e2e), but cleanup removes every +# sbx-* VM on the cocoon root, so it cannot run beside another e2e or sandboxd. set -euo pipefail TEMPLATE=${TEMPLATE:-rt:24.04} diff --git a/scripts/egress-e2e.sh b/scripts/egress-e2e.sh index 1a26b33c3..c5c5dea97 100755 --- a/scripts/egress-e2e.sh +++ b/scripts/egress-e2e.sh @@ -19,7 +19,7 @@ REACH=127.0.0.1 BRIDGE_CFG="" NICADDR="" if [[ $NET == egress ]]; then - REACH=$(ip -4 addr show "$BRIDGE" 2>/dev/null | awk '/inet /{print $2}' | cut -d/ -f1 | head -1) + REACH=$(ip -4 addr show "$BRIDGE" 2>/dev/null | awk '/inet /{print $2}' | cut -d/ -f1 | head -1 || true) [[ -z $REACH ]] && { echo "bridge $BRIDGE has no IPv4 — egress lane needs a gateway"; exit 1; } BRIDGE_CFG="\"bridges\": [\"$BRIDGE\"]," NICADDR="${REACH%.*}.222/24" # static guest NIC (image does not DHCP), so a blocked direct egress is the nft lock diff --git a/sdk/python/cocoonsandbox/conn.py b/sdk/python/cocoonsandbox/conn.py index b26225909..2dc5c21d2 100644 --- a/sdk/python/cocoonsandbox/conn.py +++ b/sdk/python/cocoonsandbox/conn.py @@ -10,13 +10,15 @@ import time import urllib.parse from collections.abc import Callable, Iterator -from typing import Any, BinaryIO, Protocol, TypeVar +from typing import Any, BinaryIO, Protocol, TypeVar, cast from .errors import APIError, ProtocolError, SandboxError, SandboxTimeout, SilkdError from .frames import MAX_FRAME, decode_response, encode_request KEEP_ALIVE_CONNS = 8 _FD_SETSIZE = 1024 +_READ_BUFFER = 64 * 1024 +_TCP_KEEPALIVE = {"TCP_KEEPIDLE": 15, "TCP_KEEPALIVE": 15, "TCP_KEEPINTVL": 15, "TCP_KEEPCNT": 9} _CloseableT = TypeVar("_CloseableT", bound="_Closeable") @@ -194,6 +196,10 @@ def dial_agent( except OSError as exc: raise _handshake_error(f"dial {addr}", exc) from exc sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) + for opt, setting in _TCP_KEEPALIVE.items(): + if hasattr(socket, opt): + sock.setsockopt(socket.IPPROTO_TCP, getattr(socket, opt), setting) request = ( f"GET /v1/sandboxes/{sandbox_id}/agent HTTP/1.1\r\n" f"Host: {endpoint.netloc}\r\n" @@ -206,7 +212,7 @@ def dial_agent( try: if endpoint.scheme != "https": sock.settimeout(None) - return Conn(sock, sock.makefile("rb"), request, timeout) + return Conn(sock, _buffered_reader(sock), request, timeout) context = ssl_context or ssl.create_default_context() sock = context.wrap_socket(sock, server_hostname=host, do_handshake_on_connect=False) sock.settimeout(remaining_timeout(timeout, deadline)) @@ -217,7 +223,7 @@ def dial_agent( # a TLS edge may be a proxy that drops bytes sent ahead of the 101 sock.settimeout(remaining_timeout(timeout, deadline)) sock.sendall(request) - reader = sock.makefile("rb") + reader = _buffered_reader(sock) _read_upgrade(reader, lambda: sock.settimeout(remaining_timeout(timeout, deadline))) remaining_timeout(timeout, deadline) sock.settimeout(None) @@ -255,6 +261,10 @@ def is_quiet(sock: socket.socket | None) -> bool: return False +def _buffered_reader(sock: socket.socket) -> BinaryIO: + return cast(BinaryIO, sock.makefile("rb", buffering=_READ_BUFFER)) + + def _handshake_error(what: str, exc: OSError) -> SandboxError: if isinstance(exc, (TimeoutError, socket.timeout)): return SandboxTimeout(f"{what} timed out") diff --git a/sdk/python/tests/test_hardening.py b/sdk/python/tests/test_hardening.py index 1f4bbaa6e..0b3b017cb 100644 --- a/sdk/python/tests/test_hardening.py +++ b/sdk/python/tests/test_hardening.py @@ -37,6 +37,14 @@ def test_dial_agent_wraps_refused_connection(dead_addr): dial_agent(dead_addr, "sb_1", "tok", 0.5) +def test_dial_agent_enables_tcp_keepalive(black_hole): + with dial_agent(black_hole, "sb_1", "tok", 0.5) as conn: + assert conn._sock.getsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE) + for idle in ("TCP_KEEPIDLE", "TCP_KEEPALIVE"): + if hasattr(socket, idle): + assert conn._sock.getsockopt(socket.IPPROTO_TCP, getattr(socket, idle)) == 15, idle + + def test_send_on_a_dead_socket_raises_a_typed_error(): ours, theirs = socket.socketpair() theirs.close() diff --git a/silkd/Dockerfile b/silkd/Dockerfile index 0a6d9208d..cf68dd28f 100644 --- a/silkd/Dockerfile +++ b/silkd/Dockerfile @@ -7,7 +7,7 @@ # /silkd-static is musl-static for guests without glibc, such as the android # flavor's non-systemd userspace. Built per-platform (amd64/arm64) on native # runners; TARGETARCH picks the musl target triple. -FROM rust:1.99.1-slim@sha256:b8e062b1c3a80183df83d9ba187cadd2b5ab84bc78d453dbfeb2fccd5f982247 AS build +FROM rust:1.99.1-slim-bookworm@sha256:24707e70a4f1e23d9a7b25202ec43126d28a52294b96aad5c670df1fd1f1d400 AS build ARG TARGETARCH WORKDIR /silkd # Before any rustup/cargo call: the pin, not the base image's default, is the diff --git a/silkd/src/find.rs b/silkd/src/find.rs index 643240ff8..d991c7270 100644 --- a/silkd/src/find.rs +++ b/silkd/src/find.rs @@ -1,6 +1,7 @@ //! Search verbs: a pattern travels as data, never shell-quoted through exec. use std::io::Read; +use std::os::unix::fs::OpenOptionsExt; use std::path::{Path, PathBuf}; use std::sync::Arc; @@ -20,6 +21,8 @@ const MATCH_QUEUE: usize = 256; /// Bytes of match content in flight: above one size-bound match, so a reserve never blocks forever, far below 256 of them. const MATCH_QUEUE_BYTES: usize = 2 * FIND_MAX_FILE as usize; +const LINE_ANCHORS: [&str; 4] = ["^", "$", "\\A", "\\z"]; + struct MatchBudget { bytes: Semaphore, rt: Handle, @@ -60,6 +63,7 @@ struct Walk<'a> { name_re: Option<&'a Regex>, tx: &'a mpsc::Sender, budget: &'a MatchBudget, + prefilter: bool, } impl Walk<'_> { @@ -103,7 +107,9 @@ impl Walk<'_> { /// Scans one file; an unreadable or oversized file is skipped, not an error. fn scan_file(&self, path: &Path, body: &mut String) -> bool { - if !matches!(read_bounded(path, body), Ok(true)) { + if !matches!(read_bounded(path, body), Ok(true)) + || (self.prefilter && !self.re.is_match(body)) + { return true; } let mut name: Option> = None; @@ -222,6 +228,7 @@ where Some(Ok(re)) => Some(re), Some(Err(e)) => return proto::error_frame(w, ErrorKind::BadRequest, e.to_string()).await, }; + let prefilter = !LINE_ANCHORS.iter().any(|a| pattern.contains(a)); // one blocking-pool dispatch for the whole tree, not three per file. let (tx, mut rx) = mpsc::channel::(MATCH_QUEUE); let budget = Arc::new(MatchBudget::new(budget_bytes, Handle::current())); @@ -232,6 +239,7 @@ where name_re: name_re.as_ref(), tx: &tx, budget: &walker_budget, + prefilter, } .run(PathBuf::from(path)) }); @@ -286,7 +294,10 @@ where /// Reads a file into `body` through one handle, stopping at FIND_MAX_FILE; `Ok(false)` when the file is above the bound. fn read_bounded(path: &Path, body: &mut String) -> std::io::Result { - let file = std::fs::File::open(path)?; + let file = std::fs::OpenOptions::new() + .read(true) + .custom_flags(libc::O_NONBLOCK) + .open(path)?; let len = file.metadata()?.len(); if len > FIND_MAX_FILE { return Ok(false); @@ -423,12 +434,57 @@ mod tests { name_re: None, tx: &tx, budget: &budget, + prefilter: true, }; let completed = tokio::task::block_in_place(|| walk.run(dir.path().to_path_buf())).expect("walk"); assert!(!completed); } + #[test] + fn read_bounded_does_not_wait_on_a_fifo_with_an_idle_writer() { + let dir = tempfile::tempdir().expect("tempdir"); + let fifo = dir.path().join("fifo"); + nix::unistd::mkfifo(&fifo, nix::sys::stat::Mode::S_IRWXU).expect("mkfifo"); + let _writer = std::fs::OpenOptions::new() + .read(true) + .write(true) + .open(&fifo) + .expect("open writer"); + let (tx, rx) = std::sync::mpsc::channel(); + std::thread::spawn(move || { + let _ = tx.send(read_bounded(&fifo, &mut String::new()).is_err()); + }); + let failed = rx + .recv_timeout(Duration::from_secs(5)) + .expect("read_bounded must not wait for data"); + assert!(failed); + } + + #[tokio::test(flavor = "multi_thread")] + async fn an_anchored_pattern_matches_a_line_start_inside_the_file() { + let dir = tempfile::tempdir().expect("tempdir"); + tokio::fs::write(dir.path().join("a.txt"), "bar\nfoo here\nbaz foo\n") + .await + .expect("write"); + let (mut w, reader) = sink(); + let (_peer, mut client) = silent_client(); + find_bounded( + &mut client, + &mut w, + dir.path().display().to_string(), + "^foo".into(), + None, + MATCH_QUEUE_BYTES, + ) + .await + .expect("find"); + drop(w); + let out = reader.await.expect("join"); + assert_eq!(out.matches("\"type\":\"match\"").count(), 1, "{out}"); + assert!(out.contains("\"line\":2"), "{out}"); + } + #[tokio::test(flavor = "multi_thread")] async fn find_reports_an_oversized_match_as_an_error_frame() { let dir = tempfile::tempdir().expect("tempdir"); diff --git a/silkd/src/fs.rs b/silkd/src/fs.rs index 9870f2448..b7971b80f 100644 --- a/silkd/src/fs.rs +++ b/silkd/src/fs.rs @@ -1,6 +1,6 @@ //! Filesystem verbs: the vsock boundary is the trust boundary, so paths are taken as-is. -use std::io; +use std::io::{self, Read}; use std::os::unix::fs::{MetadataExt, OpenOptionsExt, PermissionsExt}; use std::path::{Path, PathBuf}; use std::time::UNIX_EPOCH; @@ -9,7 +9,9 @@ use tokio::fs; use tokio::io::{AsyncBufRead, AsyncWrite, AsyncWriteExt}; use tokio::sync::mpsc; -use crate::proto::{self, DirEntry, ErrorKind, FileInfo, FileKind, Response, err_frame}; +use crate::proto::{ + self, BULK_CHUNK, DirEntry, ErrorKind, FileInfo, FileKind, Response, err_frame, +}; /// Entries per `entries` frame; a worst-case 1.6KiB entry keeps a full batch under MAX_FRAME. pub const LIST_BATCH: usize = 4096; @@ -56,19 +58,33 @@ pub async fn read(w: &mut W, path: String) -> io::Result< .read(true) .custom_flags(libc::O_NONBLOCK) .open(&path)?; - let regular = file.metadata()?.is_file(); - Ok::<_, io::Error>((file, regular)) + let meta = file.metadata()?; + if !meta.is_file() { + return Ok(None); + } + let mut head = Vec::with_capacity(meta.len().min(BULK_CHUNK as u64) as usize); + let read = (&file) + .take(BULK_CHUNK as u64) + .read_to_end(&mut head) + .map(|_| head); + Ok::<_, io::Error>(Some((file, read))) }) .await .map_err(io::Error::other)?; - let mut file = match opened { - Ok((file, true)) => fs::File::from_std(file), - Ok((_, false)) => { + let (file, head) = match opened { + Ok(Some((file, Ok(head)))) => (file, head), + Ok(Some((_, Err(e)))) => return err_frame(w, &e, "read").await, + Ok(None) => { return proto::error_frame(w, ErrorKind::BadRequest, "not a regular file").await; } Err(e) => return err_frame(w, &e, "open").await, }; - if let Err(e) = proto::stream_data_frames(&mut file, w).await? { + if !head.is_empty() { + proto::write_chunk_frame(w, &mut Vec::new(), "data", &head).await?; + } + if head.len() == BULK_CHUNK + && let Err(e) = proto::stream_data_frames(&mut fs::File::from_std(file), w).await? + { return err_frame(w, &e, "read").await; } proto::write_frame(w, &Response::Done).await diff --git a/silkd/src/proto.rs b/silkd/src/proto.rs index 71cfcdc1f..981ad5c04 100644 --- a/silkd/src/proto.rs +++ b/silkd/src/proto.rs @@ -28,6 +28,7 @@ pub const READ_CHUNK: usize = 32 * 1024; pub const BULK_CHUNK: usize = 256 * 1024; const DATA_FRAME_HEAD: &[u8] = b"{\"v\":1,\"op\":\"data\",\"data\":\""; +const STDIN_FRAME_HEAD: &[u8] = b"{\"v\":1,\"op\":\"stdin\",\"data\":\""; /// Client → server frames; unknown JSON fields are ignored, which is forward compatibility. #[derive(Debug, Deserialize)] @@ -566,17 +567,23 @@ pub async fn write_feed_error(w: &mut W, err: FeedError) #[inline] pub(crate) fn parse_request(frame: &[u8], buf: &mut Vec) -> serde_json::Result { - if let Some(n) = decode_data_frame(frame, buf) { + if let Some(n) = decode_bulk_frame(frame, DATA_FRAME_HEAD, buf) { buf.truncate(n); return Ok(Request::Data { data: std::mem::take(buf), }); } + if let Some(n) = decode_bulk_frame(frame, STDIN_FRAME_HEAD, buf) { + buf.truncate(n); + return Ok(Request::Stdin { + data: std::mem::take(buf), + }); + } serde_json::from_slice(frame) } -fn decode_data_frame(frame: &[u8], out: &mut Vec) -> Option { - let b64 = frame.strip_prefix(DATA_FRAME_HEAD)?.strip_suffix(b"\"}")?; +fn decode_bulk_frame(frame: &[u8], head: &[u8], out: &mut Vec) -> Option { + let b64 = frame.strip_prefix(head)?.strip_suffix(b"\"}")?; out.resize(base64::decoded_len_estimate(b64.len()), 0); STANDARD.decode_slice(b64, out).ok() } @@ -860,9 +867,44 @@ mod tests { } #[test] - fn data_frame_head_carries_the_proto_version() { + fn bulk_frame_heads_carry_the_proto_version() { let head = format!("{{\"v\":{PROTO_VERSION},"); - assert!(DATA_FRAME_HEAD.starts_with(head.as_bytes())); + for frame_head in [DATA_FRAME_HEAD, STDIN_FRAME_HEAD] { + assert!(frame_head.starts_with(head.as_bytes())); + } + } + + #[test] + fn bulk_frame_heads_match_the_fixture_shape() { + for (head, fixture) in [ + (DATA_FRAME_HEAD, "req_data.json"), + (STDIN_FRAME_HEAD, "req_stdin.json"), + ] { + let raw = std::fs::read(format!("{FIXTURES_DIR}/{fixture}")).expect("read fixture"); + assert!(raw.starts_with(head), "{fixture}"); + } + } + + #[test] + fn fast_and_serde_stdin_frames_decode_the_same_bytes() { + let payloads: [&[u8]; 4] = [b"", b"a", b"\xff\xff\xfe", &[0x5a; 1000]]; + let mut buf = Vec::new(); + for payload in payloads { + let b64 = STANDARD.encode(payload); + for line in [ + format!("{{\"v\":1,\"op\":\"stdin\",\"data\":\"{b64}\"}}"), + format!( + "{{\"v\":1,\"op\":\"stdin\",\"data\":\"{}\"}}", + b64.replace('/', "\\/") + ), + format!("{{\"op\":\"stdin\",\"data\":\"{b64}\",\"future\":1}}"), + ] { + match parse_request(line.as_bytes(), &mut buf) { + Ok(Request::Stdin { data }) => assert_eq!(data, payload, "{line}"), + other => panic!("{line}: {other:?}"), + } + } + } } #[tokio::test] diff --git a/silkd/tests/fs_e2e.rs b/silkd/tests/fs_e2e.rs index 193934ca0..fdee6b886 100644 --- a/silkd/tests/fs_e2e.rs +++ b/silkd/tests/fs_e2e.rs @@ -142,6 +142,35 @@ async fn mkdir_rename_rm_lifecycle() { assert!(!moved.exists()); } +#[tokio::test] +async fn read_streams_files_around_the_first_chunk_bound() { + let dir = tempfile::tempdir().unwrap(); + let chunk = silkd::proto::BULK_CHUNK; + for (len, frames) in [(0, 0), (chunk, 1), (chunk + 1, 2)] { + let path = dir.path().join(format!("f{len}")); + let body: Vec = (0..len).map(|i| (i % 251) as u8).collect(); + std::fs::write(&path, &body).unwrap(); + let read = exchange(&[json!({"op":"fs_read","path":path}).to_string()]).await; + let data = read.iter().filter(|f| type_of(f) == "data").count(); + assert_eq!(data, frames, "len {len}"); + assert!(payload(&read, "data") == body, "len {len}"); + assert_eq!(type_of(read.last().unwrap()), "done", "len {len}"); + } +} + +#[cfg(target_os = "linux")] +#[tokio::test] +async fn read_returns_a_proc_file_that_reports_zero_length() { + let read = exchange(&[json!({"op":"fs_read","path":"/proc/self/status"}).to_string()]).await; + assert!( + String::from_utf8(payload(&read, "data")) + .unwrap() + .contains("Name:"), + "{read:?}" + ); + assert_eq!(type_of(read.last().unwrap()), "done"); +} + #[tokio::test] async fn read_of_a_non_regular_file_is_bad_request() { let dir = tempfile::tempdir().unwrap();