From aa90773ad8f6e91fa6da959b3f0adab90dd290e2 Mon Sep 17 00:00:00 2001 From: Leynos Date: Tue, 21 Apr 2026 12:58:43 +0000 Subject: [PATCH 1/3] Mask ACP initialize capabilities in protocol mode Intercept the first newline-delimited Agentic Control Protocol (ACP) `initialize` frame from protocol stdin and remove host-delegated `terminal` and `fs` capabilities before the container sees the metadata. Bound the first-frame inspection buffer so oversized or unterminated input streams forward promptly into the container instead of waiting indefinitely. Document the ACP masking trust boundary, add unit and BDD coverage, and mark roadmap Step 2.6.1 as complete. --- .../2-6-1-intercept-acp-initialization.md | 343 +++++++++++++ docs/podbot-design.md | 7 + docs/podbot-roadmap.md | 2 +- docs/users-guide.md | 11 +- src/engine/connection/exec/protocol.rs | 131 ++++- .../connection/exec/protocol_acp_tests.rs | 458 ++++++++++++++++++ tests/features/acp_capability_masking.feature | 22 + 7 files changed, 971 insertions(+), 3 deletions(-) create mode 100644 docs/execplans/2-6-1-intercept-acp-initialization.md create mode 100644 src/engine/connection/exec/protocol_acp_tests.rs create mode 100644 tests/features/acp_capability_masking.feature diff --git a/docs/execplans/2-6-1-intercept-acp-initialization.md b/docs/execplans/2-6-1-intercept-acp-initialization.md new file mode 100644 index 00000000..99945c06 --- /dev/null +++ b/docs/execplans/2-6-1-intercept-acp-initialization.md @@ -0,0 +1,343 @@ +# Step 2.6.1: Intercept Agentic Control Protocol (ACP) initialization + +This ExecPlan (execution plan) is a living document. The sections +`Constraints`, `Tolerances`, `Risks`, `Progress`, `Surprises and discoveries`, +`Decision log`, and `Outcomes and retrospective` must be kept up to date as +work proceeds. + +Status: COMPLETE (2026-04-21) + +No `PLANS.md` file exists in this repository as of 2026-04-21, so this ExecPlan +is the governing implementation document for this task. + +## Purpose and big picture + +Complete the first implementation task under Step 2.6 from +`docs/podbot-roadmap.md`: intercept ACP initialization and mask `terminal/*` +and `fs/*` capabilities before forwarding capability metadata into the +container. + +Podbot already had a protocol-safe exec path in +`src/engine/connection/exec/protocol.rs`, but that path was intentionally a raw +byte proxy. That was correct for generic stdio protocols, yet insufficient for +ACP hosting because an IDE-side ACP client can advertise host terminal and +filesystem delegation features during `initialize`. Forwarding those +capabilities unchanged would let the hosted agent discover authority that +Podbot is meant to withhold across the sandbox boundary. + +Observable success for this task: + +- the protocol stdin proxy inspects the first newline-delimited ACP frame; +- an ACP `initialize` request has `params.clientCapabilities.terminal` and + `params.clientCapabilities.fs` removed before the bytes reach the container; +- unrelated capabilities remain intact; +- malformed or non-ACP first frames pass through unchanged so protocol + transparency remains intact for other transports; +- existing protocol-mode guarantees remain intact: no TTY, no resize + propagation, and accurate exit-code reporting; +- unit tests with `rstest` and behavioural tests with `rstest-bdd` v0.5.0 + cover happy, unhappy, and edge paths; +- `docs/podbot-design.md` and `docs/users-guide.md` reflect the shipped + behaviour; +- only the relevant Step 2.6 roadmap checkbox is marked done. + +## Constraints + +- Scope is limited to Step 2.6.1, not the full Step 2.6 feature set. Do not + implement the later runtime denylist or delegation override here. +- Preserve the existing protocol proxy contract from Step 2.5: + stdout purity, stderr-only diagnostics, bounded buffering, and explicit stdin + shutdown ordering. +- Preserve attached-mode terminal handling, including initial resize, Unix + `SIGWINCH` propagation, and current exit-code reporting. +- Do not add a new dependency. Reuse existing workspace crates and reexports. +- Keep touched Rust modules below the repository's 400-line guidance; prefer + small focused test modules over inflating an existing file. +- Use `rstest` fixtures and parameterized cases for unit coverage. +- Use `rstest-bdd` v0.5.0 for behavioural coverage. +- Keep production code panic-free; use tolerant pass-through behaviour when ACP + parsing fails. +- Use en-GB-oxendict spelling in comments and documentation. +- Run documentation gates because this change updates Markdown files: + `make fmt`, `make markdownlint`, and `make nixie`. +- Run the requested Rust gates before completion: + `make check-fmt`, `make lint`, and `make test`. + +## Tolerances (exception triggers) + +- Scope tolerance: stop and escalate if the work requires the `podbot host` + command to exist before masking can be validated. +- Interface tolerance: stop and escalate if finishing the task requires a + public API break rather than an internal proxy-seam change. +- Protocol tolerance: stop and escalate if masking generic protocol mode would + corrupt non-ACP traffic rather than safely identifying ACP `initialize`. +- Dependency tolerance: stop and escalate before adding a new crate. +- Iteration tolerance: if the full gate stack still fails after three focused + fix passes, document the blocker and escalate. + +## Risks + +- Risk: `ExecMode::Protocol` is generic, not ACP-specific. Blindly rewriting + all protocol stdin would be too broad. Severity: high. Likelihood: high. + Mitigation: only inspect the first newline-delimited frame and only rewrite a + message that actually looks like ACP `initialize` with + `params.clientCapabilities`. + +- Risk: parsing and rewriting the first frame could break the byte-preserving + proxy contract. Severity: medium. Likelihood: medium. Mitigation: limit the + rewrite to the first frame only, preserve line endings, and leave malformed + or non-ACP frames unchanged. + +- Risk: feature-file edits for `rstest-bdd` are compile-time inputs, so stale + generated code can make scenario matching look broken. Severity: medium. + Likelihood: medium. Mitigation: keep scenario titles synchronized with the + feature file and use a clean rebuild only if the generated bindings appear + stale. + +- Risk: adding ACP-specific coverage on top of the existing protocol proxy + tests could duplicate too much helper code. Severity: low. Likelihood: + medium. Mitigation: keep focused masking tests in + `src/engine/connection/exec/protocol_acp_tests.rs`. + +## Context and orientation + +Relevant design and implementation anchors: + +- `docs/podbot-roadmap.md` Step 2.6 defines ACP capability masking as the next + container-engine integration task. +- `docs/podbot-design.md` already commits Podbot to masking ACP + `terminal/*` and `fs/*` capabilities by default. +- `src/engine/connection/exec/protocol.rs` is the current library seam for + protocol-safe stdin/stdout/stderr proxying. +- `src/engine/connection/exec/mod.rs` dispatches `ExecMode::Protocol` through + that proxy and remains the single place that waits for daemon exit codes. +- `src/engine/connection/exec/attached.rs` and + `src/engine/connection/exec/terminal.rs` own interactive terminal-only + behaviour and must remain unchanged in semantics. + +Before this change, the protocol stdin path was: + +1. Wrap host stdin in a bounded `BufReader`. +2. Copy bytes directly into container stdin. +3. Flush and shut down container stdin. + +After this change, the stdin path becomes: + +1. Wrap host stdin in the existing bounded `BufReader`. +2. Read the first newline-delimited frame. +3. If it is an ACP `initialize` request with `clientCapabilities.terminal` or + `clientCapabilities.fs`, rewrite that JSON frame with those keys removed. +4. Write the first frame to container stdin. +5. Resume the normal byte-transparent copy loop for the remainder of stdin. + +This keeps the ACP-specific logic at the narrowest seam that can be tested +without waiting for the future `podbot host` command implementation. + +## Agent team execution model + +Use a small agent team so planning and implementation stay parallel but +reviewable. + +Lane A (docs and roadmap reconnaissance owner): + +- Read the roadmap, design, and testing guidance. +- Confirm what behaviour and documentation updates are required. + +Lane B (code-path reconnaissance owner): + +- Trace `ExecMode::Protocol`, terminal handling, resize propagation, and exit + code reporting. +- Confirm where ACP masking can land without disturbing interactive sessions. + +Lane C (primary implementation owner in the main thread): + +- Draft this ExecPlan. +- Implement ACP masking at the protocol proxy seam. +- Add unit and behavioural coverage. +- Update documentation and run all gates. + +Coordination rule: + +- Use the reconnaissance lanes to reduce uncertainty up front, then keep all + code edits and final integration in the main thread so the change remains + coherent and atomic. + +## Plan of work + +### Stage A: Confirm the correct landing zone + +Verify that the protocol proxy seam, not the interactive terminal path, is the +correct place to intercept ACP `initialize`. + +Completed result: + +- Confirmed `ExecMode::Protocol` dispatches through + `src/engine/connection/exec/protocol.rs`. +- Confirmed attached-mode resize propagation and terminal handling are isolated + in `attached.rs` and `terminal.rs`. +- Confirmed no ACP-specific logic existed anywhere else in the codebase. + +### Stage B: Rewrite only ACP initialize capability metadata + +Add a narrow helper pipeline in `protocol.rs` that: + +- reads the first newline-delimited frame from host stdin; +- parses it as JSON if possible; +- rewrites only ACP `initialize` requests that expose + `params.clientCapabilities`; +- removes `terminal` and `fs` capability families; +- preserves unrelated capabilities and original line endings; +- forwards malformed or unrelated frames unchanged. + +Completed result: + +- Added ACP masking helpers in `src/engine/connection/exec/protocol.rs`. +- Kept the rewrite on the first frame only, then resumed the existing raw + proxy loop for the remainder of stdin. + +### Stage C: Add focused unit coverage + +Add `rstest` unit coverage at two levels: + +- direct helper tests for frame masking and line-ending preservation in + `src/engine/connection/exec/protocol_acp_tests.rs`; +- session-level proxy tests in + `src/engine/connection/exec/tests/proxy_helpers/acp_masking.rs`. + +Coverage requirements: + +- happy: ACP initialize removes `terminal` and `fs`; +- happy: unrelated capabilities remain; +- edge: empty `clientCapabilities` is removed after masking if nothing remains; +- edge: non-initialize and malformed messages are unchanged; +- happy: trailing protocol bytes after the masked frame remain unchanged; +- unhappy: input flush failures still surface correctly after ACP masking. + +### Stage D: Add behavioural coverage with `rstest-bdd` + +Add a dedicated ACP feature file and scenario bindings describing the operator- +visible behaviour in plain language. + +Completed result: + +- `tests/features/acp_capability_masking.feature` covers: + - blocked-capability masking; + - malformed initialize pass-through; + - initialize without blocked capabilities remaining unchanged. +- `src/engine/connection/exec/protocol_acp_tests.rs` binds those scenarios via + `rstest-bdd` v0.5.0. + +### Stage E: Update documentation and roadmap state + +Update: + +- `docs/podbot-design.md` to record the first-frame rewrite decision and the + unchanged pass-through rule for malformed or non-ACP frames; +- `docs/users-guide.md` to explain ACP masking within protocol mode; +- `docs/podbot-roadmap.md` to mark only the first Step 2.6 checkbox done. + +### Stage F: Validate the complete change + +Run the full gate stack with `tee` and `set -o pipefail`. + +```shell +set -o pipefail +make fmt 2>&1 | tee /tmp/podbot-make-fmt.log +make markdownlint 2>&1 | tee /tmp/podbot-make-markdownlint.log +make nixie 2>&1 | tee /tmp/podbot-make-nixie.log +make check-fmt 2>&1 | tee /tmp/podbot-make-check-fmt.log +make lint 2>&1 | tee /tmp/podbot-make-lint.log +make test 2>&1 | tee /tmp/podbot-make-test.log +``` + +## Validation and acceptance + +The task is complete only when all of the following are true: + +- ACP `initialize` capability metadata is masked before reaching the + containerised agent; +- malformed and non-ACP first frames remain unchanged; +- protocol stdout purity, non-TTY behaviour, and exit-code reporting remain + intact; +- `rstest` and `rstest-bdd` cover happy, unhappy, and edge paths; +- `docs/podbot-design.md` and `docs/users-guide.md` document the final + behaviour; +- only the relevant roadmap checkbox is marked done; +- the full documentation and Rust gate stack passes. + +## Idempotence and recovery + +- The masking logic is additive and local to `protocol.rs`. If interrupted, + revert only that seam rather than touching interactive exec code. +- If `.feature` edits appear to be ignored, do a clean rebuild before retrying + tests. +- Keep future Step 2.6 work separate: runtime denylist and explicit delegation + overrides should land in follow-on changes rather than stretching this one. + +## Progress + +- [x] (2026-04-21) Retrieved project memory, loaded repo guidance, and reviewed + roadmap/design/testing documents. +- [x] (2026-04-21) Used a lane-based agent-team plan inside the ExecPlan to + separate documentation, code-path, implementation, and validation work. +- [x] (2026-04-21) Confirmed `protocol.rs` is the correct seam for ACP + initialization masking. +- [x] (2026-04-21) Implemented first-frame ACP masking in + `src/engine/connection/exec/protocol.rs`. +- [x] (2026-04-21) Added focused `rstest` coverage in + `src/engine/connection/exec/protocol_acp_tests.rs`. +- [x] (2026-04-21) Added `rstest-bdd` coverage in + `tests/features/acp_capability_masking.feature`. +- [x] (2026-04-21) Updated `docs/podbot-design.md`, `docs/users-guide.md`, and + `docs/podbot-roadmap.md`. +- [x] (2026-04-21) Ran `make fmt`, `make markdownlint`, `make nixie`, + `make check-fmt`, `make lint`, and `make test` successfully. + +## Surprises and discoveries + +- Discovery: `rstest-bdd` feature files are compile-time inputs. Scenario-name + mismatches can reflect stale generated code rather than wrong Rust bindings, + so a clean rebuild is the recovery step if they ever diverge. +- Discovery: Step 2.6 can be delivered meaningfully before `podbot host` + exists by treating the protocol proxy as the library-owned ACP seam. + +## Decision log + +- Decision: keep masking in `src/engine/connection/exec/protocol.rs` instead + of waiting for `podbot host`. Rationale: that is the narrowest seam through + which ACP bytes already flow, and it is fully testable now. + +- Decision: rewrite only the first newline-delimited frame. Rationale: ACP + initialization occurs once at the start of the session, and limiting the + parser scope preserves the existing raw-proxy contract for later traffic. + +- Decision: treat masking failure as transparent pass-through, not a hard + protocol error. Rationale: Step 2.6.1 is about removing known capabilities, + not blocking or diagnosing later ACP misuse. A malformed first frame might + belong to another protocol, and guessing would be riskier than forwarding it. + +- Decision: remove `params.clientCapabilities` entirely when masking leaves it + empty. Rationale: ACP treats omitted capabilities as unsupported, so dropping + an empty object is semantically accurate and keeps the forwarded frame tidy. + +- Decision: mark only the first Step 2.6 roadmap checkbox done. Rationale: the + runtime denylist, stderr denials, and explicit override are still future work. + +## Outcomes and retrospective + +- Shipped: ACP capability masking now occurs before the container sees the + hosted client's `initialize` request. +- Shipped: protocol sessions still preserve non-TTY execution, avoid resize + handling, and report exit codes through the existing inspect loop. +- Shipped: happy, unhappy, and edge coverage now exists at both unit and + behavioural levels. +- Documentation: the design document and users guide now describe the actual + masking behaviour, including unchanged forwarding for malformed or unrelated + first frames. +- Roadmap: the Step 2.6 checkbox for intercepting ACP initialization is now + marked done, while later Step 2.6 work remains open. +- Gates passed: `make fmt`, `make markdownlint`, `make nixie`, + `make check-fmt`, `make lint`, and `make test`. +- Remaining follow-up: implement the runtime denylist, protocol errors for + blocked methods, and an explicit delegation override in subsequent Step 2.6 + changes. diff --git a/docs/podbot-design.md b/docs/podbot-design.md index b365650b..bacae44d 100644 --- a/docs/podbot-design.md +++ b/docs/podbot-design.md @@ -212,6 +212,13 @@ ACP masking is an implementation requirement, not a documentation preference: - Podbot must parse the ACP initialization handshake and remove blocked capability families before forwarding capability data to the hosted agent. +- The current implementation performs that masking by reading the first + newline-delimited ACP frame from host stdin, rewriting an `initialize` + request when `params.clientCapabilities` advertises `terminal` or `fs`, and + then resuming the normal raw byte proxy for the rest of the session. +- If the first protocol frame is malformed or not an ACP `initialize` request, + Podbot forwards it unchanged rather than guessing another protocol's + semantics. - Podbot must maintain a runtime denylist for blocked ACP methods and return a protocol error if those methods are attempted later in the session. - The delegation override must be explicit, disabled by default, and surfaced diff --git a/docs/podbot-roadmap.md b/docs/podbot-roadmap.md index 4b620b0f..491acbac 100644 --- a/docs/podbot-roadmap.md +++ b/docs/podbot-roadmap.md @@ -189,7 +189,7 @@ Prevent ACP client-side delegation from bypassing container sandbox boundaries. **Tasks:** -- [ ] Intercept ACP initialization and mask `terminal/*` and `fs/*` capabilities +- [x] Intercept ACP initialization and mask `terminal/*` and `fs/*` capabilities by default before forwarding capability metadata. - [ ] Enforce a runtime denylist for blocked ACP methods after initialization. - [ ] Return protocol errors for blocked methods and record denials on stderr. diff --git a/docs/users-guide.md b/docs/users-guide.md index 2cf72d79..5ac50283 100644 --- a/docs/users-guide.md +++ b/docs/users-guide.md @@ -63,6 +63,11 @@ This subcommand is temporarily unavailable in the current release. If you run session. Use `podbot run` or the library exec API for now; the release notes will call out when hosted mode becomes available. +When hosted mode uses ACP, podbot masks host-delegated `terminal/*` and `fs/*` +capabilities from the initial ACP `initialize` request before forwarding it to +the sandboxed agent. This keeps the default trust boundary aligned with the +container sandbox even when the hosting client advertises broader ACP support. + | Option | Required | Default | Description | | -------------- | -------- | --------------- | ------------------------------------------ | | `--agent` | No | Config/defaults | Agent type: `claude`, `codex`, or `custom` | @@ -124,7 +129,11 @@ Execution behaviour: host stderr without terminal framing or interactive echo injection. Protocol mode uses bounded buffering (64 KiB buffers for stdin forwarding and output chunks) so hosted protocols can apply backpressure; if the host stdout writer - blocks, the proxy yields and backpressure propagates to the container. + blocks, the proxy yields and backpressure propagates to the container. When + the proxied protocol is ACP, podbot rewrites the first `initialize` request + to remove `clientCapabilities.terminal` and `clientCapabilities.fs` before + the bytes reach the container. If that first frame is malformed or belongs to + another protocol, podbot forwards it unchanged. - TTY allocation is enabled only when attached mode is selected and both local stdin and stdout are terminals. - When TTY is enabled, podbot sends an initial resize to the daemon. On Unix diff --git a/src/engine/connection/exec/protocol.rs b/src/engine/connection/exec/protocol.rs index 6f9c9efb..6f70fb8c 100644 --- a/src/engine/connection/exec/protocol.rs +++ b/src/engine/connection/exec/protocol.rs @@ -28,7 +28,8 @@ use std::time::Duration; use bollard::container::LogOutput; use bollard::errors::Error as BollardError; use futures_util::{Stream, StreamExt}; -use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt}; +use ortho_config::serde_json::{self, Value}; +use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt}; use tokio::task::JoinHandle; use tokio::time::timeout; @@ -59,6 +60,10 @@ const STDIN_SETTLE_TIMEOUT: Duration = Duration::from_millis(50); /// protocol message sizes and typical OS pipe buffer defaults, keeping latency /// low while preventing unbounded accumulation during high-throughput scenarios. const STDIN_BUFFER_CAPACITY: usize = 65_536; +const ACP_INITIALIZE_METHOD: &str = "initialize"; +const ACP_CLIENT_CAPABILITIES_FIELD: &str = "clientCapabilities"; +const ACP_FILE_SYSTEM_CAPABILITY: &str = "fs"; +const ACP_TERMINAL_CAPABILITY: &str = "terminal"; #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub(super) struct ProtocolSessionOptions { @@ -238,11 +243,135 @@ where HostStdin: AsyncRead + Unpin, { let mut buffered_stdin = tokio::io::BufReader::with_capacity(STDIN_BUFFER_CAPACITY, host_stdin); + forward_initial_protocol_frame_async(&mut buffered_stdin, &mut input).await?; tokio::io::copy(&mut buffered_stdin, &mut input).await?; input.flush().await?; input.shutdown().await } +async fn forward_initial_protocol_frame_async( + buffered_stdin: &mut tokio::io::BufReader, + input: &mut Pin>, +) -> io::Result<()> +where + HostStdin: AsyncRead + Unpin, +{ + let mut first_frame = Vec::new(); + loop { + if first_frame.len() == STDIN_BUFFER_CAPACITY { + input.write_all(&first_frame).await?; + return Ok(()); + } + + let (bytes_to_consume, has_complete_frame) = { + let buffered = buffered_stdin.fill_buf().await?; + if buffered.is_empty() { + return forward_partial_initial_frame(input, &first_frame).await; + } + + let remaining_capacity = STDIN_BUFFER_CAPACITY.saturating_sub(first_frame.len()); + let bytes_to_scan = buffered.len().min(remaining_capacity); + let newline_offset = buffered + .get(..bytes_to_scan) + .and_then(|bytes| bytes.iter().position(|byte| *byte == b'\n')); + let bytes_to_consume = newline_offset.map_or(bytes_to_scan, |offset| offset + 1); + let Some(bytes) = buffered.get(..bytes_to_consume) else { + return Err(io::Error::other( + "buffered stdin slice exceeded available input", + )); + }; + first_frame.extend_from_slice(bytes); + (bytes_to_consume, newline_offset.is_some()) + }; + + buffered_stdin.consume(bytes_to_consume); + + if has_complete_frame { + return input + .write_all(&mask_acp_initialize_frame(&first_frame)) + .await; + } + } +} + +async fn forward_partial_initial_frame( + input: &mut Pin>, + first_frame: &[u8], +) -> io::Result<()> { + if first_frame.is_empty() { + return Ok(()); + } + + input.write_all(first_frame).await +} + +fn mask_acp_initialize_frame(frame: &[u8]) -> Vec { + let (payload, line_ending) = split_frame_line_ending(frame); + + let Ok(mut message) = serde_json::from_slice(payload) else { + return frame.to_vec(); + }; + + if !remove_masked_acp_capabilities(&mut message) { + return frame.to_vec(); + } + + let Ok(mut serialized) = serde_json::to_vec(&message) else { + return frame.to_vec(); + }; + serialized.extend_from_slice(line_ending); + serialized +} + +fn split_frame_line_ending(frame: &[u8]) -> (&[u8], &[u8]) { + if let Some(stripped) = frame.strip_suffix(b"\r\n") { + return (stripped, b"\r\n"); + } + + if let Some(stripped) = frame.strip_suffix(b"\n") { + return (stripped, b"\n"); + } + + (frame, b"") +} + +fn remove_masked_acp_capabilities(message: &mut Value) -> bool { + if message.get("method").and_then(Value::as_str) != Some(ACP_INITIALIZE_METHOD) { + return false; + } + + let Some(params) = message.get_mut("params").and_then(Value::as_object_mut) else { + return false; + }; + + let Some(client_capabilities) = params + .get_mut(ACP_CLIENT_CAPABILITIES_FIELD) + .and_then(Value::as_object_mut) + else { + return false; + }; + + let removed_terminal = client_capabilities + .remove(ACP_TERMINAL_CAPABILITY) + .is_some(); + let removed_fs = client_capabilities + .remove(ACP_FILE_SYSTEM_CAPABILITY) + .is_some(); + if !removed_terminal && !removed_fs { + return false; + } + + if client_capabilities.is_empty() { + params.remove(ACP_CLIENT_CAPABILITIES_FIELD); + } + + true +} + +#[cfg(test)] +#[path = "protocol_acp_tests.rs"] +mod acp_tests; + async fn run_output_loop_async( container_id: &str, output: &mut Pin> + Send>>, diff --git a/src/engine/connection/exec/protocol_acp_tests.rs b/src/engine/connection/exec/protocol_acp_tests.rs new file mode 100644 index 00000000..1ba5b1a7 --- /dev/null +++ b/src/engine/connection/exec/protocol_acp_tests.rs @@ -0,0 +1,458 @@ +//! ACP capability masking tests for the protocol stdin proxy. + +use std::io; +use std::pin::Pin; +use std::sync::{Arc, Mutex}; +use std::task::{Context, Poll}; + +use rstest::rstest; +use rstest_bdd::Slot; +use rstest_bdd_macros::{ScenarioState, given, scenario, then, when}; +use tokio::io::{AsyncWrite, AsyncWriteExt, DuplexStream}; + +use super::*; + +type StepResult = Result; + +#[derive(Default, ScenarioState)] +struct AcpMaskingState { + host_stdin: Slot>, + expected_forwarded: Slot>, + actual_forwarded: Slot>, + succeeded: Slot, +} + +struct RecordingInputWriter { + bytes: Arc>>, + shutdown_called: Arc>, +} + +impl RecordingInputWriter { + fn new() -> Self { + Self { + bytes: Arc::new(Mutex::new(Vec::new())), + shutdown_called: Arc::new(Mutex::new(false)), + } + } +} + +impl AsyncWrite for RecordingInputWriter { + fn poll_write( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + self.bytes + .lock() + .expect("writer mutex should not poison") + .extend_from_slice(buf); + Poll::Ready(Ok(buf.len())) + } + + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + *self + .shutdown_called + .lock() + .expect("shutdown mutex should not poison") = true; + Poll::Ready(Ok(())) + } +} + +fn initialize_frame(line_ending: &str) -> Vec { + let payload = serde_json::json!({ + "jsonrpc": "2.0", + "id": 0, + "method": "initialize", + "params": { + "protocolVersion": 1, + "clientCapabilities": { + "fs": { + "readTextFile": true, + "writeTextFile": true + }, + "terminal": true, + "_meta": { + "custom": true + } + }, + "clientInfo": { + "name": "podbot-tests", + "version": "1.0.0" + } + } + }); + + let mut frame = serde_json::to_vec(&payload).expect("initialize payload should serialize"); + frame.extend_from_slice(line_ending.as_bytes()); + frame +} + +fn initialize_without_blocked_capabilities() -> Vec { + let payload = serde_json::json!({ + "jsonrpc": "2.0", + "id": 0, + "method": "initialize", + "params": { + "protocolVersion": 1, + "clientCapabilities": { + "_meta": { + "custom": true + } + } + } + }); + + let mut frame = serde_json::to_vec(&payload).expect("initialize payload should serialize"); + frame.push(b'\n'); + frame +} + +fn initialize_with_only_blocked_capabilities(line_ending: &str) -> Vec { + let payload = serde_json::json!({ + "jsonrpc": "2.0", + "id": 0, + "method": "initialize", + "params": { + "protocolVersion": 1, + "clientCapabilities": { + "fs": { + "readTextFile": true, + "writeTextFile": true + }, + "terminal": true + }, + "clientInfo": { + "name": "podbot-tests", + "version": "1.0.0" + } + } + }); + + let mut frame = serde_json::to_vec(&payload).expect("initialize payload should serialize"); + frame.extend_from_slice(line_ending.as_bytes()); + frame +} + +fn session_new_bytes() -> Vec { + br#"{"jsonrpc":"2.0","id":1,"method":"session/new","params":{"cwd":"/workspace"}}"#.to_vec() +} + +fn malformed_initialize_bytes() -> Vec { + br#"{"jsonrpc":"2.0","method":"initialize","params":{"clientCapabilities":{"terminal":true}}"# + .to_vec() +} + +fn parse_frame_payload(frame: &[u8]) -> serde_json::Value { + let (payload, _) = split_frame_line_ending(frame); + serde_json::from_slice(payload).expect("frame should contain JSON payload") +} + +fn assert_masked_client_capabilities(message: &serde_json::Value) { + let client_capabilities = message + .get("params") + .and_then(serde_json::Value::as_object) + .and_then(|params| params.get("clientCapabilities")) + .and_then(serde_json::Value::as_object) + .expect("clientCapabilities should remain present"); + assert!( + !client_capabilities.contains_key(ACP_FILE_SYSTEM_CAPABILITY), + "fs capability should be removed" + ); + assert!( + !client_capabilities.contains_key(ACP_TERMINAL_CAPABILITY), + "terminal capability should be removed" + ); + assert!( + client_capabilities.contains_key("_meta"), + "unrelated capabilities should remain" + ); +} + +async fn build_host_stdin(bytes: &[u8]) -> io::Result { + let capacity = bytes.len().max(1); + let (mut writer, reader) = tokio::io::duplex(capacity); + writer.write_all(bytes).await?; + drop(writer); + Ok(reader) +} + +fn run_forwarding(host_stdin_bytes: &[u8]) -> (Vec, bool) { + let runtime = tokio::runtime::Runtime::new().expect("runtime should build"); + let host_stdin = runtime + .block_on(build_host_stdin(host_stdin_bytes)) + .expect("host stdin should build"); + let container_input = RecordingInputWriter::new(); + let forwarded_bytes = container_input.bytes.clone(); + let shutdown_called = container_input.shutdown_called.clone(); + + runtime + .block_on(forward_host_stdin_to_exec_async( + host_stdin, + Box::pin(container_input), + )) + .expect("stdin forwarding should succeed"); + + ( + forwarded_bytes + .lock() + .expect("writer mutex should not poison") + .clone(), + *shutdown_called + .lock() + .expect("shutdown mutex should not poison"), + ) +} + +#[rstest] +#[case("\n")] +#[case("\r\n")] +fn mask_acp_initialize_frame_removes_blocked_capabilities(#[case] line_ending: &str) { + let frame = initialize_frame(line_ending); + let masked = mask_acp_initialize_frame(&frame); + let payload = parse_frame_payload(&masked); + + assert_eq!( + split_frame_line_ending(&masked).1, + line_ending.as_bytes(), + "line ending should be preserved" + ); + assert_masked_client_capabilities(&payload); +} + +#[test] +fn mask_acp_initialize_frame_removes_empty_client_capabilities() { + let frame = initialize_with_only_blocked_capabilities("\n"); + let masked = mask_acp_initialize_frame(&frame); + let payload = parse_frame_payload(&masked); + let params = payload + .get("params") + .and_then(serde_json::Value::as_object) + .expect("initialize params should remain present"); + + assert!( + !params.contains_key("clientCapabilities"), + "clientCapabilities should be removed when all entries are masked" + ); + assert_eq!( + params.get("protocolVersion"), + Some(&serde_json::json!(1)), + "protocolVersion should remain unchanged" + ); + assert_eq!( + params.get("clientInfo"), + Some(&serde_json::json!({ + "name": "podbot-tests", + "version": "1.0.0" + })), + "clientInfo should remain unchanged" + ); +} + +#[test] +fn mask_acp_initialize_frame_leaves_non_initialize_messages_unchanged() { + let mut frame = session_new_bytes(); + frame.push(b'\n'); + + assert_eq!(mask_acp_initialize_frame(&frame), frame); +} + +#[test] +fn mask_acp_initialize_frame_leaves_malformed_input_unchanged() { + let frame = malformed_initialize_bytes(); + + assert_eq!(mask_acp_initialize_frame(&frame), frame); +} + +#[test] +fn forwarding_masks_initialize_and_preserves_trailing_bytes() { + let mut host_stdin_bytes = initialize_frame("\n"); + let trailing = session_new_bytes(); + host_stdin_bytes.extend_from_slice(&trailing); + + let (forwarded, shutdown_called) = run_forwarding(&host_stdin_bytes); + let newline_index = forwarded + .iter() + .position(|byte| *byte == b'\n') + .expect("masked initialize should remain line terminated"); + let initialize_frame = forwarded + .get(..=newline_index) + .expect("masked initialize frame should remain addressable"); + let trailing_forwarded = forwarded + .get(newline_index + 1..) + .expect("trailing bytes should remain addressable"); + let payload = parse_frame_payload(initialize_frame); + + assert_masked_client_capabilities(&payload); + assert_eq!( + trailing_forwarded, + trailing.as_slice(), + "trailing bytes should pass through unchanged" + ); + assert!(shutdown_called, "stdin forwarding should shut down input"); +} + +#[test] +fn forwarding_does_not_wait_indefinitely_for_oversized_initial_frame() { + let runtime = tokio::runtime::Runtime::new().expect("runtime should build"); + let host_stdin_bytes = vec![b'x'; STDIN_BUFFER_CAPACITY + 1]; + let (host_writer, host_reader) = runtime + .block_on(async { + let (mut host_writer, host_reader) = tokio::io::duplex(host_stdin_bytes.len()); + host_writer.write_all(&host_stdin_bytes).await?; + io::Result::Ok((host_writer, host_reader)) + }) + .expect("host stdin should accept oversized initial bytes"); + + let mut buffered_stdin = + tokio::io::BufReader::with_capacity(STDIN_BUFFER_CAPACITY, host_reader); + let recording_input = RecordingInputWriter::new(); + let forwarded_bytes = recording_input.bytes.clone(); + let mut container_input: Pin> = Box::pin(recording_input); + + runtime + .block_on(async { + tokio::time::timeout( + STDIN_SETTLE_TIMEOUT, + forward_initial_protocol_frame_async(&mut buffered_stdin, &mut container_input), + ) + .await + }) + .expect("initial forwarding should not wait for newline or EOF") + .expect("initial forwarding should succeed"); + + assert_eq!( + forwarded_bytes + .lock() + .expect("writer mutex should not poison") + .len(), + STDIN_BUFFER_CAPACITY, + "only the bounded first-frame buffer should be held before streaming resumes" + ); + + drop(host_writer); +} + +#[rstest::fixture] +fn acp_masking_state() -> AcpMaskingState { + AcpMaskingState::default() +} + +fn masked_initialize_with_follow_up() -> (Vec, Vec) { + let mut host_stdin_bytes = initialize_frame("\n"); + host_stdin_bytes.extend_from_slice(&session_new_bytes()); + + let expected_initialize = serde_json::json!({ + "jsonrpc": "2.0", + "id": 0, + "method": "initialize", + "params": { + "protocolVersion": 1, + "clientCapabilities": { + "_meta": { + "custom": true + } + }, + "clientInfo": { + "name": "podbot-tests", + "version": "1.0.0" + } + } + }); + + let mut expected = serde_json::to_vec(&expected_initialize) + .expect("expected initialize payload should serialize"); + expected.push(b'\n'); + expected.extend_from_slice(&session_new_bytes()); + + (host_stdin_bytes, expected) +} + +#[given( + "ACP stdin contains an initialize request with blocked capabilities and a follow-up request" +)] +fn acp_stdin_contains_blocked_initialize_and_follow_up(acp_masking_state: &AcpMaskingState) { + let (host_stdin_bytes, expected) = masked_initialize_with_follow_up(); + acp_masking_state.host_stdin.set(host_stdin_bytes); + acp_masking_state.expected_forwarded.set(expected); +} + +#[given("ACP stdin contains malformed initialize bytes")] +fn acp_stdin_contains_malformed_initialize(acp_masking_state: &AcpMaskingState) { + let malformed = malformed_initialize_bytes(); + acp_masking_state.host_stdin.set(malformed.clone()); + acp_masking_state.expected_forwarded.set(malformed); +} + +#[given("ACP stdin contains initialize without blocked capabilities")] +fn acp_stdin_contains_safe_initialize(acp_masking_state: &AcpMaskingState) { + let initialize = initialize_without_blocked_capabilities(); + acp_masking_state.host_stdin.set(initialize.clone()); + acp_masking_state.expected_forwarded.set(initialize); +} + +#[when("ACP stdin forwarding runs")] +fn acp_stdin_forwarding_runs(acp_masking_state: &AcpMaskingState) -> StepResult<()> { + let host_stdin_bytes = acp_masking_state + .host_stdin + .get() + .ok_or_else(|| String::from("host stdin should be configured"))?; + let (forwarded, _) = run_forwarding(&host_stdin_bytes); + acp_masking_state.actual_forwarded.set(forwarded); + acp_masking_state.succeeded.set(true); + Ok(()) +} + +#[then("ACP stdin forwarding succeeds")] +fn acp_stdin_forwarding_succeeds(acp_masking_state: &AcpMaskingState) -> StepResult<()> { + if acp_masking_state.succeeded.get() == Some(true) { + Ok(()) + } else { + Err(String::from("expected ACP stdin forwarding to succeed")) + } +} + +#[then("the forwarded ACP stdin matches the expected bytes")] +fn forwarded_acp_stdin_matches_expected(acp_masking_state: &AcpMaskingState) -> StepResult<()> { + let actual = acp_masking_state + .actual_forwarded + .get() + .ok_or_else(|| String::from("forwarded bytes should be recorded"))?; + let expected = acp_masking_state + .expected_forwarded + .get() + .ok_or_else(|| String::from("expected bytes should be recorded"))?; + + if actual == expected { + Ok(()) + } else { + Err(format!("expected {expected:?}, got {actual:?}")) + } +} + +#[scenario( + path = "tests/features/acp_capability_masking.feature", + name = "ACP initialize masks blocked capabilities before forwarding" +)] +fn acp_initialize_masks_blocked_capabilities(acp_masking_state: AcpMaskingState) { + let _ = acp_masking_state; +} + +#[scenario( + path = "tests/features/acp_capability_masking.feature", + name = "Malformed ACP initialize is forwarded unchanged" +)] +fn malformed_acp_initialize_is_forwarded_unchanged(acp_masking_state: AcpMaskingState) { + let _ = acp_masking_state; +} + +#[scenario( + path = "tests/features/acp_capability_masking.feature", + name = "ACP initialize without blocked capabilities stays unchanged" +)] +fn acp_initialize_without_blocked_capabilities_stays_unchanged(acp_masking_state: AcpMaskingState) { + let _ = acp_masking_state; +} diff --git a/tests/features/acp_capability_masking.feature b/tests/features/acp_capability_masking.feature new file mode 100644 index 00000000..32e82ee7 --- /dev/null +++ b/tests/features/acp_capability_masking.feature @@ -0,0 +1,22 @@ +Feature: ACP capability masking + + ACP hosting must remove host-delegated terminal and filesystem capabilities + from the initialize request before the request reaches the sandboxed agent. + + Scenario: ACP initialize masks blocked capabilities before forwarding + Given ACP stdin contains an initialize request with blocked capabilities and a follow-up request + When ACP stdin forwarding runs + Then ACP stdin forwarding succeeds + And the forwarded ACP stdin matches the expected bytes + + Scenario: Malformed ACP initialize is forwarded unchanged + Given ACP stdin contains malformed initialize bytes + When ACP stdin forwarding runs + Then ACP stdin forwarding succeeds + And the forwarded ACP stdin matches the expected bytes + + Scenario: ACP initialize without blocked capabilities stays unchanged + Given ACP stdin contains initialize without blocked capabilities + When ACP stdin forwarding runs + Then ACP stdin forwarding succeeds + And the forwarded ACP stdin matches the expected bytes From 5a391bccd9cdb1987f77701aa48dfce133766ffd Mon Sep 17 00:00:00 2001 From: leynos Date: Thu, 30 Apr 2026 18:09:19 +0200 Subject: [PATCH 2/3] Address ACP protocol review findings Move ACP initialize masking helpers out of the protocol proxy module so the proxy loop stays focused and under the file-length budget. Keep ACP first-frame rewriting behind a protocol-session option so generic protocol sessions retain raw stdin forwarding semantics unless ACP masking is selected. Split ACP BDD scenario wiring out of the unit-test module, extend coverage for the first-frame-only contract, and document the current hosted-mode ACP masking scope while `podbot host` remains forthcoming. --- docs/developers-guide.md | 9 + .../2-6-1-intercept-acp-initialization.md | 28 ++-- docs/users-guide.md | 11 +- src/engine/connection/exec/acp_helpers.rs | 136 +++++++++++++++ src/engine/connection/exec/protocol.rs | 156 +++--------------- .../connection/exec/protocol_acp_bdd_tests.rs | 110 ++++++++++++ .../connection/exec/protocol_acp_tests.rs | 150 +++++------------ src/engine/connection/exec/session.rs | 31 ++++ 8 files changed, 374 insertions(+), 257 deletions(-) create mode 100644 src/engine/connection/exec/acp_helpers.rs create mode 100644 src/engine/connection/exec/protocol_acp_bdd_tests.rs diff --git a/docs/developers-guide.md b/docs/developers-guide.md index e9a50aa1..7eb841e1 100644 --- a/docs/developers-guide.md +++ b/docs/developers-guide.md @@ -357,6 +357,15 @@ When adding a new execution mode: - `ContainerExecClient` mock implementations for unit testing without a live daemon. +#### 8.2.1. ACP initialization rewrite contract + +When protocol tests wrap `run_session` with `RecordingWriter` handles, they +must preserve the ACP initialization rewrite contract: inspect only the first +newline-delimited frame, rewrite only ACP `initialize` requests that expose +`params.clientCapabilities`, and remove the `terminal` and `fs` capability +families before the frame reaches the container. Preserve the original line +endings, and let malformed or non-ACP frames pass through unchanged. + ### 8.3. Parameterized tests Use `#[rstest(...)]` to eliminate duplicated test cases. Group related diff --git a/docs/execplans/2-6-1-intercept-acp-initialization.md b/docs/execplans/2-6-1-intercept-acp-initialization.md index 99945c06..9304e633 100644 --- a/docs/execplans/2-6-1-intercept-acp-initialization.md +++ b/docs/execplans/2-6-1-intercept-acp-initialization.md @@ -179,7 +179,8 @@ Completed result: ### Stage B: Rewrite only ACP initialize capability metadata -Add a narrow helper pipeline in `protocol.rs` that: +Add a narrow helper pipeline in +`src/engine/connection/exec/acp_helpers.rs` that: - reads the first newline-delimited frame from host stdin; - parses it as JSON if possible; @@ -191,7 +192,8 @@ Add a narrow helper pipeline in `protocol.rs` that: Completed result: -- Added ACP masking helpers in `src/engine/connection/exec/protocol.rs`. +- Added ACP masking helpers in + `src/engine/connection/exec/acp_helpers.rs`. - Kept the rewrite on the first frame only, then resumed the existing raw proxy loop for the remainder of stdin. @@ -224,8 +226,8 @@ Completed result: - blocked-capability masking; - malformed initialize pass-through; - initialize without blocked capabilities remaining unchanged. -- `src/engine/connection/exec/protocol_acp_tests.rs` binds those scenarios via - `rstest-bdd` v0.5.0. +- `src/engine/connection/exec/protocol_acp_bdd_tests.rs` binds those scenarios + via `rstest-bdd` v0.5.0. ### Stage E: Update documentation and roadmap state @@ -255,7 +257,7 @@ make test 2>&1 | tee /tmp/podbot-make-test.log The task is complete only when all of the following are true: - ACP `initialize` capability metadata is masked before reaching the - containerised agent; + containerized agent; - malformed and non-ACP first frames remain unchanged; - protocol stdout purity, non-TTY behaviour, and exit-code reporting remain intact; @@ -282,12 +284,13 @@ The task is complete only when all of the following are true: separate documentation, code-path, implementation, and validation work. - [x] (2026-04-21) Confirmed `protocol.rs` is the correct seam for ACP initialization masking. -- [x] (2026-04-21) Implemented first-frame ACP masking in - `src/engine/connection/exec/protocol.rs`. +- [x] (2026-04-21) Implemented first-frame ACP masking helpers in + `src/engine/connection/exec/acp_helpers.rs`. - [x] (2026-04-21) Added focused `rstest` coverage in `src/engine/connection/exec/protocol_acp_tests.rs`. - [x] (2026-04-21) Added `rstest-bdd` coverage in - `tests/features/acp_capability_masking.feature`. + `tests/features/acp_capability_masking.feature` with bindings in + `src/engine/connection/exec/protocol_acp_bdd_tests.rs`. - [x] (2026-04-21) Updated `docs/podbot-design.md`, `docs/users-guide.md`, and `docs/podbot-roadmap.md`. - [x] (2026-04-21) Ran `make fmt`, `make markdownlint`, `make nixie`, @@ -303,9 +306,12 @@ The task is complete only when all of the following are true: ## Decision log -- Decision: keep masking in `src/engine/connection/exec/protocol.rs` instead - of waiting for `podbot host`. Rationale: that is the narrowest seam through - which ACP bytes already flow, and it is fully testable now. +- Decision: keep the ACP interception seam in + `src/engine/connection/exec/protocol.rs` while moving masking helpers into + `src/engine/connection/exec/acp_helpers.rs`, instead of waiting for + `podbot host`. Rationale: the protocol proxy is the narrowest seam through + which ACP bytes already flow, and the helper module keeps that seam fully + testable without overloading the proxy loop. - Decision: rewrite only the first newline-delimited frame. Rationale: ACP initialization occurs once at the start of the session, and limiting the diff --git a/docs/users-guide.md b/docs/users-guide.md index 5ac50283..663a40ef 100644 --- a/docs/users-guide.md +++ b/docs/users-guide.md @@ -63,10 +63,13 @@ This subcommand is temporarily unavailable in the current release. If you run session. Use `podbot run` or the library exec API for now; the release notes will call out when hosted mode becomes available. -When hosted mode uses ACP, podbot masks host-delegated `terminal/*` and `fs/*` -capabilities from the initial ACP `initialize` request before forwarding it to -the sandboxed agent. This keeps the default trust boundary aligned with the -container sandbox even when the hosting client advertises broader ACP support. +When hosted mode uses ACP, podbot masks `terminal/*` and `fs/*` capabilities +from the initial ACP `initialize` request before forwarding it to the +sandboxed agent. This masking currently applies only to the +protocol/library path used by hosted mode and ACP until `podbot host` is +implemented. The scope and behaviour may change when `podbot host` ships, but +for now it keeps the default trust boundary aligned with the container +sandbox even when the hosting client advertises broader ACP support. | Option | Required | Default | Description | | -------------- | -------- | --------------- | ------------------------------------------ | diff --git a/src/engine/connection/exec/acp_helpers.rs b/src/engine/connection/exec/acp_helpers.rs new file mode 100644 index 00000000..44d252c8 --- /dev/null +++ b/src/engine/connection/exec/acp_helpers.rs @@ -0,0 +1,136 @@ +//! ACP initialization frame rewriting for protocol-mode stdin forwarding. + +use std::io; +use std::pin::Pin; + +use ortho_config::serde_json::{self, Value}; +use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt}; + +use super::STDIN_BUFFER_CAPACITY; + +pub(super) const MAX_FIRST_FRAME_BYTES: usize = STDIN_BUFFER_CAPACITY; +pub(super) const ACP_INITIALIZE_METHOD: &str = "initialize"; +pub(super) const ACP_CLIENT_CAPABILITIES_FIELD: &str = "clientCapabilities"; +pub(super) const ACP_FILE_SYSTEM_CAPABILITY: &str = "fs"; +pub(super) const ACP_TERMINAL_CAPABILITY: &str = "terminal"; + +pub(super) async fn forward_initial_acp_frame_async( + buffered_stdin: &mut tokio::io::BufReader, + input: &mut Pin>, +) -> io::Result<()> +where + HostStdin: AsyncRead + Unpin, +{ + let mut first_frame = Vec::new(); + loop { + if first_frame.len() == MAX_FIRST_FRAME_BYTES { + input.write_all(&first_frame).await?; + return Ok(()); + } + + let (bytes_to_consume, has_complete_frame) = + read_next_bounded_frame_chunk(buffered_stdin, &mut first_frame).await?; + if bytes_to_consume == 0 { + input.write_all(&first_frame).await?; + return Ok(()); + } + + buffered_stdin.consume(bytes_to_consume); + + if has_complete_frame { + return input + .write_all(&mask_acp_initialize_frame(&first_frame)) + .await; + } + } +} + +async fn read_next_bounded_frame_chunk( + buffered_stdin: &mut tokio::io::BufReader, + first_frame: &mut Vec, +) -> io::Result<(usize, bool)> +where + HostStdin: AsyncRead + Unpin, +{ + let buffered = buffered_stdin.fill_buf().await?; + if buffered.is_empty() { + return Ok((0, false)); + } + + let remaining_capacity = MAX_FIRST_FRAME_BYTES.saturating_sub(first_frame.len()); + let bytes_to_scan = buffered.len().min(remaining_capacity); + let newline_offset = buffered + .get(..bytes_to_scan) + .and_then(|bytes| bytes.iter().position(|byte| *byte == b'\n')); + let bytes_to_consume = newline_offset.map_or(bytes_to_scan, |offset| offset + 1); + let Some(bytes) = buffered.get(..bytes_to_consume) else { + return Err(io::Error::other( + "buffered stdin slice exceeded available input", + )); + }; + first_frame.extend_from_slice(bytes); + Ok((bytes_to_consume, newline_offset.is_some())) +} + +pub(super) fn mask_acp_initialize_frame(frame: &[u8]) -> Vec { + let (payload, line_ending) = split_frame_line_ending(frame); + + let Ok(mut message) = serde_json::from_slice(payload) else { + return frame.to_vec(); + }; + + if !remove_masked_acp_capabilities(&mut message) { + return frame.to_vec(); + } + + let Ok(mut serialized) = serde_json::to_vec(&message) else { + return frame.to_vec(); + }; + serialized.extend_from_slice(line_ending); + serialized +} + +pub(super) fn split_frame_line_ending(frame: &[u8]) -> (&[u8], &[u8]) { + if let Some(stripped) = frame.strip_suffix(b"\r\n") { + return (stripped, b"\r\n"); + } + + if let Some(stripped) = frame.strip_suffix(b"\n") { + return (stripped, b"\n"); + } + + (frame, b"") +} + +fn remove_masked_acp_capabilities(message: &mut Value) -> bool { + if message.get("method").and_then(Value::as_str) != Some(ACP_INITIALIZE_METHOD) { + return false; + } + + let Some(params) = message.get_mut("params").and_then(Value::as_object_mut) else { + return false; + }; + + let Some(client_capabilities) = params + .get_mut(ACP_CLIENT_CAPABILITIES_FIELD) + .and_then(Value::as_object_mut) + else { + return false; + }; + + let removed_terminal = client_capabilities + .remove(ACP_TERMINAL_CAPABILITY) + .is_some(); + let removed_fs = client_capabilities + .remove(ACP_FILE_SYSTEM_CAPABILITY) + .is_some(); + if !removed_terminal && !removed_fs { + return false; + } + + if client_capabilities.is_empty() { + params.remove(ACP_CLIENT_CAPABILITIES_FIELD); + } + + true +} diff --git a/src/engine/connection/exec/protocol.rs b/src/engine/connection/exec/protocol.rs index 6f70fb8c..dabecc95 100644 --- a/src/engine/connection/exec/protocol.rs +++ b/src/engine/connection/exec/protocol.rs @@ -28,11 +28,13 @@ use std::time::Duration; use bollard::container::LogOutput; use bollard::errors::Error as BollardError; use futures_util::{Stream, StreamExt}; -use ortho_config::serde_json::{self, Value}; -use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt}; +use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt}; use tokio::task::JoinHandle; use tokio::time::timeout; +#[path = "acp_helpers.rs"] +mod acp_helpers; + use super::helpers::spawn_stdin_forwarding_task; use super::host_io::stdin_forwarding_disabled_for_tests; use super::{ExecRequest, exec_failed}; @@ -60,20 +62,18 @@ const STDIN_SETTLE_TIMEOUT: Duration = Duration::from_millis(50); /// protocol message sizes and typical OS pipe buffer defaults, keeping latency /// low while preventing unbounded accumulation during high-throughput scenarios. const STDIN_BUFFER_CAPACITY: usize = 65_536; -const ACP_INITIALIZE_METHOD: &str = "initialize"; -const ACP_CLIENT_CAPABILITIES_FIELD: &str = "clientCapabilities"; -const ACP_FILE_SYSTEM_CAPABILITY: &str = "fs"; -const ACP_TERMINAL_CAPABILITY: &str = "terminal"; #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub(super) struct ProtocolSessionOptions { disable_stdin_forwarding: bool, + rewrite_acp_initialize: bool, } impl ProtocolSessionOptions { pub(super) const fn new() -> Self { Self { disable_stdin_forwarding: false, + rewrite_acp_initialize: false, } } @@ -81,6 +81,11 @@ impl ProtocolSessionOptions { self.disable_stdin_forwarding = disable; self } + + pub(super) const fn with_acp_initialize_rewrite_enabled(mut self, enable: bool) -> Self { + self.rewrite_acp_initialize = enable; + self + } } struct HeldOpenStdin { @@ -168,8 +173,11 @@ where stderr: mut host_stderr, options, } = stdio; + let rewrite_acp_initialize = options.rewrite_acp_initialize; let stdin_task = - spawn_stdin_forwarding_task(host_stdin, input, forward_host_stdin_to_exec_async); + spawn_stdin_forwarding_task(host_stdin, input, move |stdin_reader, exec_input| { + forward_host_stdin_to_exec_async(stdin_reader, exec_input, rewrite_acp_initialize) + }); let output_result = run_output_loop_async( request.container_id(), &mut output, @@ -236,136 +244,24 @@ fn abort_stdin_forwarding_task(stdin_task: JoinHandle>) { } async fn forward_host_stdin_to_exec_async( - host_stdin: HostStdin, + mut host_stdin: HostStdin, mut input: Pin>, + rewrite_acp_initialize: bool, ) -> io::Result<()> where HostStdin: AsyncRead + Unpin, { - let mut buffered_stdin = tokio::io::BufReader::with_capacity(STDIN_BUFFER_CAPACITY, host_stdin); - forward_initial_protocol_frame_async(&mut buffered_stdin, &mut input).await?; - tokio::io::copy(&mut buffered_stdin, &mut input).await?; - input.flush().await?; - input.shutdown().await -} - -async fn forward_initial_protocol_frame_async( - buffered_stdin: &mut tokio::io::BufReader, - input: &mut Pin>, -) -> io::Result<()> -where - HostStdin: AsyncRead + Unpin, -{ - let mut first_frame = Vec::new(); - loop { - if first_frame.len() == STDIN_BUFFER_CAPACITY { - input.write_all(&first_frame).await?; - return Ok(()); - } - - let (bytes_to_consume, has_complete_frame) = { - let buffered = buffered_stdin.fill_buf().await?; - if buffered.is_empty() { - return forward_partial_initial_frame(input, &first_frame).await; - } - - let remaining_capacity = STDIN_BUFFER_CAPACITY.saturating_sub(first_frame.len()); - let bytes_to_scan = buffered.len().min(remaining_capacity); - let newline_offset = buffered - .get(..bytes_to_scan) - .and_then(|bytes| bytes.iter().position(|byte| *byte == b'\n')); - let bytes_to_consume = newline_offset.map_or(bytes_to_scan, |offset| offset + 1); - let Some(bytes) = buffered.get(..bytes_to_consume) else { - return Err(io::Error::other( - "buffered stdin slice exceeded available input", - )); - }; - first_frame.extend_from_slice(bytes); - (bytes_to_consume, newline_offset.is_some()) - }; - - buffered_stdin.consume(bytes_to_consume); - - if has_complete_frame { - return input - .write_all(&mask_acp_initialize_frame(&first_frame)) - .await; - } - } -} - -async fn forward_partial_initial_frame( - input: &mut Pin>, - first_frame: &[u8], -) -> io::Result<()> { - if first_frame.is_empty() { - return Ok(()); - } - - input.write_all(first_frame).await -} - -fn mask_acp_initialize_frame(frame: &[u8]) -> Vec { - let (payload, line_ending) = split_frame_line_ending(frame); - - let Ok(mut message) = serde_json::from_slice(payload) else { - return frame.to_vec(); - }; - - if !remove_masked_acp_capabilities(&mut message) { - return frame.to_vec(); - } - - let Ok(mut serialized) = serde_json::to_vec(&message) else { - return frame.to_vec(); - }; - serialized.extend_from_slice(line_ending); - serialized -} - -fn split_frame_line_ending(frame: &[u8]) -> (&[u8], &[u8]) { - if let Some(stripped) = frame.strip_suffix(b"\r\n") { - return (stripped, b"\r\n"); - } - - if let Some(stripped) = frame.strip_suffix(b"\n") { - return (stripped, b"\n"); - } - - (frame, b"") -} - -fn remove_masked_acp_capabilities(message: &mut Value) -> bool { - if message.get("method").and_then(Value::as_str) != Some(ACP_INITIALIZE_METHOD) { - return false; - } - - let Some(params) = message.get_mut("params").and_then(Value::as_object_mut) else { - return false; - }; - - let Some(client_capabilities) = params - .get_mut(ACP_CLIENT_CAPABILITIES_FIELD) - .and_then(Value::as_object_mut) - else { - return false; - }; - - let removed_terminal = client_capabilities - .remove(ACP_TERMINAL_CAPABILITY) - .is_some(); - let removed_fs = client_capabilities - .remove(ACP_FILE_SYSTEM_CAPABILITY) - .is_some(); - if !removed_terminal && !removed_fs { - return false; - } - - if client_capabilities.is_empty() { - params.remove(ACP_CLIENT_CAPABILITIES_FIELD); + if rewrite_acp_initialize { + let mut buffered_stdin = + tokio::io::BufReader::with_capacity(STDIN_BUFFER_CAPACITY, host_stdin); + acp_helpers::forward_initial_acp_frame_async(&mut buffered_stdin, &mut input).await?; + tokio::io::copy(&mut buffered_stdin, &mut input).await?; + } else { + tokio::io::copy(&mut host_stdin, &mut input).await?; } - true + input.flush().await?; + input.shutdown().await } #[cfg(test)] diff --git a/src/engine/connection/exec/protocol_acp_bdd_tests.rs b/src/engine/connection/exec/protocol_acp_bdd_tests.rs new file mode 100644 index 00000000..deafefbf --- /dev/null +++ b/src/engine/connection/exec/protocol_acp_bdd_tests.rs @@ -0,0 +1,110 @@ +//! BDD scenario wiring for ACP capability masking tests. + +use rstest_bdd::Slot; +use rstest_bdd_macros::{ScenarioState, given, scenario, then, when}; + +use super::{ + initialize_without_blocked_capabilities, malformed_initialize_bytes, + masked_initialize_with_follow_up, run_forwarding, +}; + +type StepResult = Result; + +#[derive(Default, ScenarioState)] +struct AcpMaskingState { + host_stdin: Slot>, + expected_forwarded: Slot>, + actual_forwarded: Slot>, + succeeded: Slot, +} + +#[rstest::fixture] +fn acp_masking_state() -> AcpMaskingState { + AcpMaskingState::default() +} + +#[given( + "ACP stdin contains an initialize request with blocked capabilities and a follow-up request" +)] +fn acp_stdin_contains_blocked_initialize_and_follow_up(acp_masking_state: &AcpMaskingState) { + let (host_stdin_bytes, expected) = masked_initialize_with_follow_up(); + acp_masking_state.host_stdin.set(host_stdin_bytes); + acp_masking_state.expected_forwarded.set(expected); +} + +#[given("ACP stdin contains malformed initialize bytes")] +fn acp_stdin_contains_malformed_initialize(acp_masking_state: &AcpMaskingState) { + let malformed = malformed_initialize_bytes(); + acp_masking_state.host_stdin.set(malformed.clone()); + acp_masking_state.expected_forwarded.set(malformed); +} + +#[given("ACP stdin contains initialize without blocked capabilities")] +fn acp_stdin_contains_safe_initialize(acp_masking_state: &AcpMaskingState) { + let initialize = initialize_without_blocked_capabilities(); + acp_masking_state.host_stdin.set(initialize.clone()); + acp_masking_state.expected_forwarded.set(initialize); +} + +#[when("ACP stdin forwarding runs")] +fn acp_stdin_forwarding_runs(acp_masking_state: &AcpMaskingState) -> StepResult<()> { + let host_stdin_bytes = acp_masking_state + .host_stdin + .get() + .ok_or_else(|| String::from("host stdin should be configured"))?; + let (forwarded, _) = run_forwarding(&host_stdin_bytes); + acp_masking_state.actual_forwarded.set(forwarded); + acp_masking_state.succeeded.set(true); + Ok(()) +} + +#[then("ACP stdin forwarding succeeds")] +fn acp_stdin_forwarding_succeeds(acp_masking_state: &AcpMaskingState) -> StepResult<()> { + if acp_masking_state.succeeded.get() == Some(true) { + Ok(()) + } else { + Err(String::from("expected ACP stdin forwarding to succeed")) + } +} + +#[then("the forwarded ACP stdin matches the expected bytes")] +fn forwarded_acp_stdin_matches_expected(acp_masking_state: &AcpMaskingState) -> StepResult<()> { + let actual = acp_masking_state + .actual_forwarded + .get() + .ok_or_else(|| String::from("forwarded bytes should be recorded"))?; + let expected = acp_masking_state + .expected_forwarded + .get() + .ok_or_else(|| String::from("expected bytes should be recorded"))?; + + if actual == expected { + Ok(()) + } else { + Err(format!("expected {expected:?}, got {actual:?}")) + } +} + +#[scenario( + path = "tests/features/acp_capability_masking.feature", + name = "ACP initialize masks blocked capabilities before forwarding" +)] +fn acp_initialize_masks_blocked_capabilities(acp_masking_state: AcpMaskingState) { + let _ = acp_masking_state; +} + +#[scenario( + path = "tests/features/acp_capability_masking.feature", + name = "Malformed ACP initialize is forwarded unchanged" +)] +fn malformed_acp_initialize_is_forwarded_unchanged(acp_masking_state: AcpMaskingState) { + let _ = acp_masking_state; +} + +#[scenario( + path = "tests/features/acp_capability_masking.feature", + name = "ACP initialize without blocked capabilities stays unchanged" +)] +fn acp_initialize_without_blocked_capabilities_stays_unchanged(acp_masking_state: AcpMaskingState) { + let _ = acp_masking_state; +} diff --git a/src/engine/connection/exec/protocol_acp_tests.rs b/src/engine/connection/exec/protocol_acp_tests.rs index 1ba5b1a7..72ffc94b 100644 --- a/src/engine/connection/exec/protocol_acp_tests.rs +++ b/src/engine/connection/exec/protocol_acp_tests.rs @@ -6,22 +6,14 @@ use std::sync::{Arc, Mutex}; use std::task::{Context, Poll}; use rstest::rstest; -use rstest_bdd::Slot; -use rstest_bdd_macros::{ScenarioState, given, scenario, then, when}; use tokio::io::{AsyncWrite, AsyncWriteExt, DuplexStream}; +use super::acp_helpers::{ + ACP_FILE_SYSTEM_CAPABILITY, ACP_TERMINAL_CAPABILITY, MAX_FIRST_FRAME_BYTES, + forward_initial_acp_frame_async, mask_acp_initialize_frame, split_frame_line_ending, +}; use super::*; -type StepResult = Result; - -#[derive(Default, ScenarioState)] -struct AcpMaskingState { - host_stdin: Slot>, - expected_forwarded: Slot>, - actual_forwarded: Slot>, - succeeded: Slot, -} - struct RecordingInputWriter { bytes: Arc>>, shutdown_called: Arc>, @@ -91,7 +83,7 @@ fn initialize_frame(line_ending: &str) -> Vec { frame } -fn initialize_without_blocked_capabilities() -> Vec { +pub(super) fn initialize_without_blocked_capabilities() -> Vec { let payload = serde_json::json!({ "jsonrpc": "2.0", "id": 0, @@ -141,7 +133,7 @@ fn session_new_bytes() -> Vec { br#"{"jsonrpc":"2.0","id":1,"method":"session/new","params":{"cwd":"/workspace"}}"#.to_vec() } -fn malformed_initialize_bytes() -> Vec { +pub(super) fn malformed_initialize_bytes() -> Vec { br#"{"jsonrpc":"2.0","method":"initialize","params":{"clientCapabilities":{"terminal":true}}"# .to_vec() } @@ -180,7 +172,14 @@ async fn build_host_stdin(bytes: &[u8]) -> io::Result { Ok(reader) } -fn run_forwarding(host_stdin_bytes: &[u8]) -> (Vec, bool) { +pub(super) fn run_forwarding(host_stdin_bytes: &[u8]) -> (Vec, bool) { + run_forwarding_with_rewrite(host_stdin_bytes, true) +} + +fn run_forwarding_with_rewrite( + host_stdin_bytes: &[u8], + rewrite_acp_initialize: bool, +) -> (Vec, bool) { let runtime = tokio::runtime::Runtime::new().expect("runtime should build"); let host_stdin = runtime .block_on(build_host_stdin(host_stdin_bytes)) @@ -193,6 +192,7 @@ fn run_forwarding(host_stdin_bytes: &[u8]) -> (Vec, bool) { .block_on(forward_host_stdin_to_exec_async( host_stdin, Box::pin(container_input), + rewrite_acp_initialize, )) .expect("stdin forwarding should succeed"); @@ -207,6 +207,19 @@ fn run_forwarding(host_stdin_bytes: &[u8]) -> (Vec, bool) { ) } +#[test] +fn forwarding_leaves_initialize_unchanged_when_acp_rewrite_is_disabled() { + let host_stdin_bytes = initialize_frame("\n"); + + let (forwarded, shutdown_called) = run_forwarding_with_rewrite(&host_stdin_bytes, false); + + assert_eq!( + forwarded, host_stdin_bytes, + "generic protocol sessions should retain raw byte-stream semantics" + ); + assert!(shutdown_called, "stdin forwarding should shut down input"); +} + #[rstest] #[case("\n")] #[case("\r\n")] @@ -270,7 +283,7 @@ fn mask_acp_initialize_frame_leaves_malformed_input_unchanged() { #[test] fn forwarding_masks_initialize_and_preserves_trailing_bytes() { let mut host_stdin_bytes = initialize_frame("\n"); - let trailing = session_new_bytes(); + let trailing = initialize_frame("\n"); host_stdin_bytes.extend_from_slice(&trailing); let (forwarded, shutdown_called) = run_forwarding(&host_stdin_bytes); @@ -298,7 +311,7 @@ fn forwarding_masks_initialize_and_preserves_trailing_bytes() { #[test] fn forwarding_does_not_wait_indefinitely_for_oversized_initial_frame() { let runtime = tokio::runtime::Runtime::new().expect("runtime should build"); - let host_stdin_bytes = vec![b'x'; STDIN_BUFFER_CAPACITY + 1]; + let host_stdin_bytes = vec![b'x'; MAX_FIRST_FRAME_BYTES + 1]; let (host_writer, host_reader) = runtime .block_on(async { let (mut host_writer, host_reader) = tokio::io::duplex(host_stdin_bytes.len()); @@ -317,7 +330,7 @@ fn forwarding_does_not_wait_indefinitely_for_oversized_initial_frame() { .block_on(async { tokio::time::timeout( STDIN_SETTLE_TIMEOUT, - forward_initial_protocol_frame_async(&mut buffered_stdin, &mut container_input), + forward_initial_acp_frame_async(&mut buffered_stdin, &mut container_input), ) .await }) @@ -329,21 +342,17 @@ fn forwarding_does_not_wait_indefinitely_for_oversized_initial_frame() { .lock() .expect("writer mutex should not poison") .len(), - STDIN_BUFFER_CAPACITY, + MAX_FIRST_FRAME_BYTES, "only the bounded first-frame buffer should be held before streaming resumes" ); drop(host_writer); } -#[rstest::fixture] -fn acp_masking_state() -> AcpMaskingState { - AcpMaskingState::default() -} - -fn masked_initialize_with_follow_up() -> (Vec, Vec) { +pub(super) fn masked_initialize_with_follow_up() -> (Vec, Vec) { let mut host_stdin_bytes = initialize_frame("\n"); - host_stdin_bytes.extend_from_slice(&session_new_bytes()); + let follow_up = initialize_frame("\n"); + host_stdin_bytes.extend_from_slice(&follow_up); let expected_initialize = serde_json::json!({ "jsonrpc": "2.0", @@ -366,93 +375,10 @@ fn masked_initialize_with_follow_up() -> (Vec, Vec) { let mut expected = serde_json::to_vec(&expected_initialize) .expect("expected initialize payload should serialize"); expected.push(b'\n'); - expected.extend_from_slice(&session_new_bytes()); + expected.extend_from_slice(&follow_up); (host_stdin_bytes, expected) } -#[given( - "ACP stdin contains an initialize request with blocked capabilities and a follow-up request" -)] -fn acp_stdin_contains_blocked_initialize_and_follow_up(acp_masking_state: &AcpMaskingState) { - let (host_stdin_bytes, expected) = masked_initialize_with_follow_up(); - acp_masking_state.host_stdin.set(host_stdin_bytes); - acp_masking_state.expected_forwarded.set(expected); -} - -#[given("ACP stdin contains malformed initialize bytes")] -fn acp_stdin_contains_malformed_initialize(acp_masking_state: &AcpMaskingState) { - let malformed = malformed_initialize_bytes(); - acp_masking_state.host_stdin.set(malformed.clone()); - acp_masking_state.expected_forwarded.set(malformed); -} - -#[given("ACP stdin contains initialize without blocked capabilities")] -fn acp_stdin_contains_safe_initialize(acp_masking_state: &AcpMaskingState) { - let initialize = initialize_without_blocked_capabilities(); - acp_masking_state.host_stdin.set(initialize.clone()); - acp_masking_state.expected_forwarded.set(initialize); -} - -#[when("ACP stdin forwarding runs")] -fn acp_stdin_forwarding_runs(acp_masking_state: &AcpMaskingState) -> StepResult<()> { - let host_stdin_bytes = acp_masking_state - .host_stdin - .get() - .ok_or_else(|| String::from("host stdin should be configured"))?; - let (forwarded, _) = run_forwarding(&host_stdin_bytes); - acp_masking_state.actual_forwarded.set(forwarded); - acp_masking_state.succeeded.set(true); - Ok(()) -} - -#[then("ACP stdin forwarding succeeds")] -fn acp_stdin_forwarding_succeeds(acp_masking_state: &AcpMaskingState) -> StepResult<()> { - if acp_masking_state.succeeded.get() == Some(true) { - Ok(()) - } else { - Err(String::from("expected ACP stdin forwarding to succeed")) - } -} - -#[then("the forwarded ACP stdin matches the expected bytes")] -fn forwarded_acp_stdin_matches_expected(acp_masking_state: &AcpMaskingState) -> StepResult<()> { - let actual = acp_masking_state - .actual_forwarded - .get() - .ok_or_else(|| String::from("forwarded bytes should be recorded"))?; - let expected = acp_masking_state - .expected_forwarded - .get() - .ok_or_else(|| String::from("expected bytes should be recorded"))?; - - if actual == expected { - Ok(()) - } else { - Err(format!("expected {expected:?}, got {actual:?}")) - } -} - -#[scenario( - path = "tests/features/acp_capability_masking.feature", - name = "ACP initialize masks blocked capabilities before forwarding" -)] -fn acp_initialize_masks_blocked_capabilities(acp_masking_state: AcpMaskingState) { - let _ = acp_masking_state; -} - -#[scenario( - path = "tests/features/acp_capability_masking.feature", - name = "Malformed ACP initialize is forwarded unchanged" -)] -fn malformed_acp_initialize_is_forwarded_unchanged(acp_masking_state: AcpMaskingState) { - let _ = acp_masking_state; -} - -#[scenario( - path = "tests/features/acp_capability_masking.feature", - name = "ACP initialize without blocked capabilities stays unchanged" -)] -fn acp_initialize_without_blocked_capabilities_stays_unchanged(acp_masking_state: AcpMaskingState) { - let _ = acp_masking_state; -} +#[path = "protocol_acp_bdd_tests.rs"] +mod bdd_tests; diff --git a/src/engine/connection/exec/session.rs b/src/engine/connection/exec/session.rs index aaf8ac52..71e5fa01 100644 --- a/src/engine/connection/exec/session.rs +++ b/src/engine/connection/exec/session.rs @@ -7,6 +7,7 @@ use super::protocol::ProtocolSessionOptions; #[derive(Debug, Clone, Copy, Default)] pub(crate) struct ExecSessionOptions { disable_protocol_stdin_forwarding: bool, + rewrite_acp_initialize: bool, } impl ExecSessionOptions { @@ -15,6 +16,7 @@ impl ExecSessionOptions { pub const fn new() -> Self { Self { disable_protocol_stdin_forwarding: false, + rewrite_acp_initialize: false, } } @@ -26,6 +28,20 @@ impl ExecSessionOptions { self.disable_protocol_stdin_forwarding = disable; self } + + /// Enable ACP initialization rewriting for protocol-mode sessions. + #[cfg_attr( + not(test), + expect( + dead_code, + reason = "reserved for production ACP session selection when podbot host is enabled" + ) + )] + #[must_use] + pub const fn with_acp_initialize_rewrite_enabled(mut self, enable: bool) -> Self { + self.rewrite_acp_initialize = enable; + self + } } pub(super) const fn protocol_session_options( @@ -33,6 +49,7 @@ pub(super) const fn protocol_session_options( ) -> ProtocolSessionOptions { ProtocolSessionOptions::new() .with_stdin_forwarding_disabled(options.disable_protocol_stdin_forwarding) + .with_acp_initialize_rewrite_enabled(options.rewrite_acp_initialize) } #[cfg(test)] @@ -46,6 +63,10 @@ mod tests { !opts.disable_protocol_stdin_forwarding, "stdin forwarding should be enabled by default", ); + assert!( + !opts.rewrite_acp_initialize, + "ACP initialize rewriting should be opt-in", + ); } #[test] @@ -88,4 +109,14 @@ mod tests { ProtocolSessionOptions::new().with_stdin_forwarding_disabled(true), ); } + + #[test] + fn protocol_session_options_reflects_acp_rewrite_flag() { + let opts = ExecSessionOptions::new().with_acp_initialize_rewrite_enabled(true); + + assert_eq!( + protocol_session_options(opts), + ProtocolSessionOptions::new().with_acp_initialize_rewrite_enabled(true), + ); + } } From 8e7c448bb93f3f72c8eefa571318b9de6a692974 Mon Sep 17 00:00:00 2001 From: "coderabbitai[bot]" <136622811+coderabbitai[bot]@users.noreply.github.com> Date: Fri, 1 May 2026 08:47:53 +0000 Subject: [PATCH 3/3] =?UTF-8?q?=F0=9F=93=9D=20CodeRabbit=20Chat:=20Impleme?= =?UTF-8?q?nt=20requested=20code=20changes?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/developers-guide.md | 2 +- docs/users-guide.md | 2 +- src/engine/connection/exec/acp_helpers.rs | 57 ++++++++++++++++++- .../connection/exec/protocol_acp_tests.rs | 18 +++++- 4 files changed, 75 insertions(+), 4 deletions(-) diff --git a/docs/developers-guide.md b/docs/developers-guide.md index 7eb841e1..56b73e6d 100644 --- a/docs/developers-guide.md +++ b/docs/developers-guide.md @@ -357,7 +357,7 @@ When adding a new execution mode: - `ContainerExecClient` mock implementations for unit testing without a live daemon. -#### 8.2.1. ACP initialization rewrite contract +#### 8.2.1. Agentic Control Protocol (ACP) initialization rewrite contract When protocol tests wrap `run_session` with `RecordingWriter` handles, they must preserve the ACP initialization rewrite contract: inspect only the first diff --git a/docs/users-guide.md b/docs/users-guide.md index 663a40ef..dbf33da6 100644 --- a/docs/users-guide.md +++ b/docs/users-guide.md @@ -63,7 +63,7 @@ This subcommand is temporarily unavailable in the current release. If you run session. Use `podbot run` or the library exec API for now; the release notes will call out when hosted mode becomes available. -When hosted mode uses ACP, podbot masks `terminal/*` and `fs/*` capabilities +When hosted mode uses Agentic Control Protocol (ACP), podbot masks `terminal/*` and `fs/*` capabilities from the initial ACP `initialize` request before forwarding it to the sandboxed agent. This masking currently applies only to the protocol/library path used by hosted mode and ACP until `podbot host` is diff --git a/src/engine/connection/exec/acp_helpers.rs b/src/engine/connection/exec/acp_helpers.rs index 44d252c8..ac466614 100644 --- a/src/engine/connection/exec/acp_helpers.rs +++ b/src/engine/connection/exec/acp_helpers.rs @@ -1,4 +1,10 @@ -//! ACP initialization frame rewriting for protocol-mode stdin forwarding. +//! Agentic Control Protocol (ACP) initialization frame rewriting for protocol-mode stdin +//! forwarding. +//! +//! This module reads the first newline-delimited frame from host stdin, rewrites the JSON +//! payload when it is an ACP `initialize` request that advertises `terminal` or `fs` +//! capabilities, and forwards the result to the container input. All other frames — including +//! malformed or non-ACP ones — are forwarded unchanged. use std::io; use std::pin::Pin; @@ -8,12 +14,33 @@ use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt}; use super::STDIN_BUFFER_CAPACITY; +/// Maximum number of bytes buffered when reading the initial ACP frame. +/// +/// Matches `STDIN_BUFFER_CAPACITY` so the first-frame buffer never exceeds the existing +/// bounded-buffering guarantee. pub(super) const MAX_FIRST_FRAME_BYTES: usize = STDIN_BUFFER_CAPACITY; + +/// The ACP method name that triggers capability masking. pub(super) const ACP_INITIALIZE_METHOD: &str = "initialize"; + +/// The JSON field within `params` that advertises client capabilities. pub(super) const ACP_CLIENT_CAPABILITIES_FIELD: &str = "clientCapabilities"; + +/// The `fs` capability family key removed from `clientCapabilities` during masking. pub(super) const ACP_FILE_SYSTEM_CAPABILITY: &str = "fs"; + +/// The `terminal` capability family key removed from `clientCapabilities` during masking. pub(super) const ACP_TERMINAL_CAPABILITY: &str = "terminal"; +/// Reads and forwards the first newline-delimited frame from `buffered_stdin` to `input`. +/// +/// When the frame is a valid ACP `initialize` request, `terminal` and `fs` capability entries +/// are removed from `params.clientCapabilities` before the bytes reach the container. If +/// `clientCapabilities` becomes empty after removal, the field itself is dropped. +/// +/// Malformed JSON or frames that do not match the ACP `initialize` shape are forwarded without +/// modification. When no newline is found within [`MAX_FIRST_FRAME_BYTES`], the buffered bytes +/// are forwarded as-is so the caller can resume the normal byte-transparent copy loop. pub(super) async fn forward_initial_acp_frame_async( buffered_stdin: &mut tokio::io::BufReader, input: &mut Pin>, @@ -24,6 +51,10 @@ where let mut first_frame = Vec::new(); loop { if first_frame.len() == MAX_FIRST_FRAME_BYTES { + tracing::debug!( + bytes = MAX_FIRST_FRAME_BYTES, + "initial ACP frame reached capacity limit; forwarding without masking", + ); input.write_all(&first_frame).await?; return Ok(()); } @@ -31,6 +62,10 @@ where let (bytes_to_consume, has_complete_frame) = read_next_bounded_frame_chunk(buffered_stdin, &mut first_frame).await?; if bytes_to_consume == 0 { + tracing::debug!( + bytes = first_frame.len(), + "stdin closed before newline; forwarding partial initial frame without masking", + ); input.write_all(&first_frame).await?; return Ok(()); } @@ -72,24 +107,44 @@ where Ok((bytes_to_consume, newline_offset.is_some())) } +/// Rewrites an ACP `initialize` frame by removing masked capability entries. +/// +/// The function splits the frame into a JSON payload and its line ending, attempts to parse +/// the payload, and — when the message is a valid ACP `initialize` request — removes +/// `terminal` and `fs` from `params.clientCapabilities`. The original line ending bytes are +/// restored after re-serialisation. +/// +/// Returns the original frame unchanged when JSON parsing fails, the message is not an ACP +/// `initialize` request, no masked capabilities are present, or re-serialisation fails. pub(super) fn mask_acp_initialize_frame(frame: &[u8]) -> Vec { let (payload, line_ending) = split_frame_line_ending(frame); let Ok(mut message) = serde_json::from_slice(payload) else { + tracing::warn!( + bytes = frame.len(), + "ACP initialize frame is not valid JSON; forwarding unchanged", + ); return frame.to_vec(); }; if !remove_masked_acp_capabilities(&mut message) { + tracing::debug!("frame is not an ACP initialize request; forwarding unchanged"); return frame.to_vec(); } let Ok(mut serialized) = serde_json::to_vec(&message) else { + tracing::warn!("failed to serialise masked ACP initialize frame; forwarding unchanged"); return frame.to_vec(); }; + + tracing::debug!("ACP initialize frame masked: terminal and fs capabilities removed"); serialized.extend_from_slice(line_ending); serialized } +/// Splits a frame into its JSON payload and trailing line-ending bytes. +/// +/// Returns `(payload, line_ending)` where `line_ending` is `b"\r\n"`, `b"\n"`, or `b""`. pub(super) fn split_frame_line_ending(frame: &[u8]) -> (&[u8], &[u8]) { if let Some(stripped) = frame.strip_suffix(b"\r\n") { return (stripped, b"\r\n"); diff --git a/src/engine/connection/exec/protocol_acp_tests.rs b/src/engine/connection/exec/protocol_acp_tests.rs index 72ffc94b..74f63e3f 100644 --- a/src/engine/connection/exec/protocol_acp_tests.rs +++ b/src/engine/connection/exec/protocol_acp_tests.rs @@ -83,6 +83,9 @@ fn initialize_frame(line_ending: &str) -> Vec { frame } +/// Returns a serialised ACP `initialize` frame whose `clientCapabilities` contains only +/// entries that are not masked (e.g. `_meta`), so the forwarded bytes should be identical +/// to the input. pub(super) fn initialize_without_blocked_capabilities() -> Vec { let payload = serde_json::json!({ "jsonrpc": "2.0", @@ -133,6 +136,8 @@ fn session_new_bytes() -> Vec { br#"{"jsonrpc":"2.0","id":1,"method":"session/new","params":{"cwd":"/workspace"}}"#.to_vec() } +/// Returns bytes that look like an ACP `initialize` message but are not valid JSON (the +/// closing brace is missing). Used to exercise the malformed-frame pass-through path. pub(super) fn malformed_initialize_bytes() -> Vec { br#"{"jsonrpc":"2.0","method":"initialize","params":{"clientCapabilities":{"terminal":true}}"# .to_vec() @@ -172,6 +177,11 @@ async fn build_host_stdin(bytes: &[u8]) -> io::Result { Ok(reader) } +/// Runs the full ACP-enabled stdin forwarding pipeline synchronously and returns +/// `(forwarded_bytes, shutdown_called)`. +/// +/// Used by BDD scenario steps that need to drive `forward_host_stdin_to_exec_async` with +/// in-memory byte streams rather than real file descriptors. pub(super) fn run_forwarding(host_stdin_bytes: &[u8]) -> (Vec, bool) { run_forwarding_with_rewrite(host_stdin_bytes, true) } @@ -349,10 +359,16 @@ fn forwarding_does_not_wait_indefinitely_for_oversized_initial_frame() { drop(host_writer); } +/// Returns `(host_stdin_bytes, expected_forwarded_bytes)` for an ACP session that begins +/// with a full `initialize` frame (containing blocked capabilities) followed by a second +/// frame. +/// +/// The expected output has the blocked capabilities removed from the first frame while the +/// second frame is preserved byte-for-byte. pub(super) fn masked_initialize_with_follow_up() -> (Vec, Vec) { let mut host_stdin_bytes = initialize_frame("\n"); let follow_up = initialize_frame("\n"); - host_stdin_bytes.extend_from_slice(&follow_up); + let expected_initialize = serde_json::json!({ "jsonrpc": "2.0",