diff --git a/packages/paperclip-runner/docs/phase-02-local-protocol.md b/packages/paperclip-runner/docs/phase-02-local-protocol.md index bf4877461a2..261da3ad0a0 100644 --- a/packages/paperclip-runner/docs/phase-02-local-protocol.md +++ b/packages/paperclip-runner/docs/phase-02-local-protocol.md @@ -69,9 +69,13 @@ model keys, and AgentMail keys are not forwarded. ## Bounded diagnostics -The runner keeps only the configured number of lines and bytes. It reports the -retained lines, retained bytes, and dropped-line count with `harness.exited`. -This log tail is separate from canonical events and the semantic result. +The supervisor frames harness stdout and stderr with a 64 KiB per-line default +before it creates a `String`; a newline-free or otherwise oversized stdout +frame is a protocol violation and fails the run. Oversized stderr frames are +drained with bounded memory and represented by a short diagnostic marker. The +separate log-tail buffer then keeps only the configured number of lines and +bytes. It reports retained lines, retained bytes, and the dropped-line count +with `harness.exited`. ## Result and terminal authority @@ -85,6 +89,16 @@ A scripted error can report useful yielded work and exit 7. An interruption can report yielded work and exit 130. Neither process fact silently replaces the semantic result. +The fake harness terminal object is only a proposal. The runner preserves it as +`harnessTerminalProposal`, validates its enum fields, and derives the canonical +turn/run terminal states from runner-owned facts: the process exit, whether a +semantic result was observed, an accepted interrupt/cancel command, controller +closure, and protocol violations. `reconciliation` records the decision and +whether the proposal contradicted those facts. A success proposal followed by a +non-zero exit therefore becomes a failed canonical terminal; the reverse +contradiction becomes succeeded only when both the process and semantic result +succeeded. + ## Browser live mode The Vite middleware starts the same TypeScript mock-core controller used by @@ -96,6 +110,34 @@ snapshots. The browser reuses package-local shadcn-style Button, Badge, Card, and Textarea components. All visual values remain in `styles.css`. +### Local transport trust model + +The browser server is a developer-only loopback transport, not an authenticated +Paperclip API. Vite dev and preview default to `127.0.0.1`. Every +`/api/phase2/*` request is checked before route lookup or process startup: + +- both ends of the accepted socket and the HTTP `Host` must be loopback; +- an `Origin`, when a browser supplies one, must exactly match the request + origin, and cross-site Fetch Metadata values are rejected; +- JSON actions require `application/json` and bodies are capped at 16 KiB, + including streamed/chunked requests; +- the server permits at most 4 active runs, retains at most 16 total runs (old + completed entries are evicted), and permits at most 8 live stream clients per + run. + +These are secure defaults plus defense in depth: even if Vite is accidentally +started with a broad `--host`, non-loopback Phase 2 API traffic is rejected. +There is intentionally no user identity, session cookie, or reusable bearer +credential in this spike. + +Residual risk: another process running as the same local user is inside this +trust boundary and can make loopback requests. The scripted Phase 2 event set is +finite, but the retained event history inside one active run is not yet a +general-purpose per-session byte budget. A later real harness/network phase must +add authenticated capabilities, per-session event/byte budgets, durable +backpressure, and production authorization instead of widening this local +transport. + ## Deferred work Phase 2 has no durable outbox, ACK, reconnect, runner restart recovery, real diff --git a/packages/paperclip-runner/protocol/fixtures/phase-02/scripts/oversized-line.json b/packages/paperclip-runner/protocol/fixtures/phase-02/scripts/oversized-line.json new file mode 100644 index 00000000000..38776bcdec2 --- /dev/null +++ b/packages/paperclip-runner/protocol/fixtures/phase-02/scripts/oversized-line.json @@ -0,0 +1,12 @@ +{ + "schema": "paperclip.fake_harness.script.v1", + "name": "oversized-line", + "steps": [ + { + "kind": "log", + "stream": "stdout", + "text": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx" + }, + { "kind": "exit", "code": 0 } + ] +} diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs b/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs index 33c53fbe93e..04d61f2ecf7 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs @@ -45,6 +45,7 @@ fn run() -> Result<(), Phase2Error> { delay_override_ms: optional_u64(&args, "--delay-ms")?, log_max_lines: usize_value(&args, "--log-max-lines", 32)?, log_max_bytes: usize_value(&args, "--log-max-bytes", 16_384)?, + harness_max_line_bytes: usize_value(&args, "--harness-max-line-bytes", 64 * 1024)?, shutdown_grace: Duration::from_millis( optional_u64(&args, "--shutdown-grace-ms")?.unwrap_or(100), ), diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/phase2.rs b/packages/paperclip-runner/runner/crates/runner-core/src/phase2.rs index 89ab5295de7..34bb9bcb551 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/phase2.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/phase2.rs @@ -1,7 +1,7 @@ use std::collections::{HashMap, VecDeque}; use std::error::Error; use std::fmt::{self, Display, Formatter}; -use std::io::{BufRead, BufReader, Write}; +use std::io::{self, BufRead, BufReader, Write}; use std::path::{Path, PathBuf}; use std::process::{Child, ChildStdin, Command, ExitStatus, Stdio}; use std::sync::mpsc::{self, Receiver, RecvTimeoutError}; @@ -46,6 +46,7 @@ pub struct RunnerConfig { pub delay_override_ms: Option, pub log_max_lines: usize, pub log_max_bytes: usize, + pub harness_max_line_bytes: usize, pub shutdown_grace: Duration, } @@ -134,10 +135,105 @@ pub enum FakeHarnessStep { enum ProcessOutput { Stdout(String), Stderr(String), + StdoutError(String), StdoutClosed, StderrClosed, } +#[derive(Debug, PartialEq, Eq)] +enum BoundedLine { + Line(String), + TooLong, + Eof, +} + +fn read_bounded_line(reader: &mut R, max_bytes: usize) -> io::Result { + let max_bytes = max_bytes.max(1); + let mut bytes = Vec::with_capacity(max_bytes.min(8 * 1024)); + let mut too_long = false; + + loop { + let buffer = reader.fill_buf()?; + if buffer.is_empty() { + if too_long { + return Ok(BoundedLine::TooLong); + } + if bytes.is_empty() { + return Ok(BoundedLine::Eof); + } + return String::from_utf8(bytes) + .map(BoundedLine::Line) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error)); + } + + let newline = buffer.iter().position(|byte| *byte == b'\n'); + let content_len = newline.unwrap_or(buffer.len()); + if !too_long { + let remaining = max_bytes.saturating_sub(bytes.len()); + let copy_len = remaining.min(content_len); + bytes.extend_from_slice(&buffer[..copy_len]); + if content_len > remaining { + too_long = true; + bytes.clear(); + } + } + let consumed = newline.map_or(buffer.len(), |index| index + 1); + reader.consume(consumed); + + if newline.is_some() { + if too_long { + return Ok(BoundedLine::TooLong); + } + if bytes.last() == Some(&b'\r') { + bytes.pop(); + } + return String::from_utf8(bytes) + .map(BoundedLine::Line) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error)); + } + } +} + +fn forward_bounded_output( + reader: R, + sender: mpsc::Sender, + stdout: bool, + max_line_bytes: usize, +) { + thread::spawn(move || { + let mut reader = BufReader::new(reader); + loop { + let output = match read_bounded_line(&mut reader, max_line_bytes) { + Ok(BoundedLine::Line(line)) if stdout => ProcessOutput::Stdout(line), + Ok(BoundedLine::Line(line)) => ProcessOutput::Stderr(line), + Ok(BoundedLine::TooLong) if stdout => ProcessOutput::StdoutError(format!( + "harness stdout frame exceeded {max_line_bytes} bytes" + )), + Ok(BoundedLine::TooLong) => ProcessOutput::Stderr(format!( + "[harness stderr frame exceeded {max_line_bytes} bytes]" + )), + Ok(BoundedLine::Eof) => break, + Err(error) if stdout => { + ProcessOutput::StdoutError(format!("failed to read harness stdout: {error}")) + } + Err(error) => { + ProcessOutput::Stderr(format!("[failed to read harness stderr: {error}]")) + } + }; + let terminal_output = matches!(&output, ProcessOutput::StdoutError(_)); + if sender.send(output).is_err() || terminal_output { + return; + } + } + let closed = if stdout { + ProcessOutput::StdoutClosed + } else { + ProcessOutput::StderrClosed + }; + let _ = sender.send(closed); + }); +} + #[derive(Clone, Debug, Serialize)] #[serde(rename_all = "camelCase")] pub struct ProcessExitFact { @@ -218,6 +314,7 @@ impl SupervisedProcess { program: &Path, args: &[String], shutdown_grace: Duration, + max_line_bytes: usize, ) -> Result { let mut command = Command::new(program); command @@ -252,43 +349,8 @@ impl SupervisedProcess { .take() .ok_or_else(|| Phase2Error::invalid("supervised process stderr was not piped"))?; let (sender, output) = mpsc::channel(); - let stdout_sender = sender.clone(); - thread::spawn(move || { - for line in BufReader::new(stdout).lines() { - match line { - Ok(line) => { - if stdout_sender.send(ProcessOutput::Stdout(line)).is_err() { - return; - } - } - Err(error) => { - let _ = stdout_sender.send(ProcessOutput::Stderr(format!( - "failed to read harness stdout: {error}" - ))); - break; - } - } - } - let _ = stdout_sender.send(ProcessOutput::StdoutClosed); - }); - thread::spawn(move || { - for line in BufReader::new(stderr).lines() { - match line { - Ok(line) => { - if sender.send(ProcessOutput::Stderr(line)).is_err() { - return; - } - } - Err(error) => { - let _ = sender.send(ProcessOutput::Stderr(format!( - "failed to read harness stderr: {error}" - ))); - break; - } - } - } - let _ = sender.send(ProcessOutput::StderrClosed); - }); + forward_bounded_output(stdout, sender.clone(), true, max_line_bytes.max(1)); + forward_bounded_output(stderr, sender, false, max_line_bytes.max(1)); Ok(Self { child, @@ -334,6 +396,9 @@ impl SupervisedProcess { match self.recv_timeout(remaining) { Ok(ProcessOutput::Stdout(line)) => return Ok(Some(line)), Ok(ProcessOutput::Stderr(_)) | Ok(ProcessOutput::StderrClosed) => {} + Ok(ProcessOutput::StdoutError(message)) => { + return Err(Phase2Error::invalid(message)); + } Ok(ProcessOutput::StdoutClosed) => return Ok(None), Err(RecvTimeoutError::Timeout) => return Ok(None), Err(RecvTimeoutError::Disconnected) => { @@ -453,6 +518,115 @@ enum RunnerStreamMessage<'a> { }, } +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +enum TerminalIntent { + #[default] + Natural, + Interrupted, + Cancelled, + ControllerClosed, + ProtocolViolation, +} + +fn enum_field<'a>(value: &'a Value, key: &str, allowed: &[&str]) -> Option<&'a str> { + value + .get(key) + .and_then(Value::as_str) + .filter(|candidate| allowed.contains(candidate)) +} + +fn valid_terminal_proposal(value: &Value) -> bool { + value.get("schema").and_then(Value::as_str) == Some("paperclip.prp.terminal.v1") + && enum_field( + value, + "turnTerminalState", + &["completed", "failed", "interrupted", "cancelled"], + ) + .is_some() + && enum_field( + value, + "runTerminalState", + &["succeeded", "failed", "cancelled"], + ) + .is_some() + && enum_field( + value, + "reportedWorkDisposition", + &["done", "blocked", "needs_review", "yielded"], + ) + .is_some() +} + +fn reconcile_terminal( + proposal: Option, + exit: &ProcessExitFact, + semantic_result: Option<&Value>, + intent: TerminalIntent, +) -> Value { + let (turn_terminal_state, run_terminal_state, reason) = match intent { + TerminalIntent::Interrupted => ("interrupted", "cancelled", "turn_interrupted"), + TerminalIntent::Cancelled => ("cancelled", "cancelled", "run_cancelled"), + TerminalIntent::ControllerClosed => ("cancelled", "cancelled", "controller_closed"), + TerminalIntent::ProtocolViolation => ("failed", "failed", "harness_protocol_violation"), + TerminalIntent::Natural if exit.success && semantic_result.is_some() => { + ("completed", "succeeded", "successful_process_and_result") + } + TerminalIntent::Natural if !exit.success => ("failed", "failed", "harness_process_failed"), + TerminalIntent::Natural => ("failed", "failed", "semantic_result_missing"), + }; + + let proposal_valid = proposal.as_ref().is_some_and(valid_terminal_proposal); + let proposal_contradicted = proposal.as_ref().is_some_and(|value| { + proposal_valid + && (value.get("turnTerminalState").and_then(Value::as_str) != Some(turn_terminal_state) + || value.get("runTerminalState").and_then(Value::as_str) + != Some(run_terminal_state)) + }); + let proposed_disposition = proposal.as_ref().and_then(|value| { + enum_field( + value, + "reportedWorkDisposition", + &["done", "blocked", "needs_review", "yielded"], + ) + .map(str::to_owned) + }); + let result_disposition = semantic_result.and_then(|value| { + enum_field( + value, + "reportedWorkDisposition", + &["done", "blocked", "needs_review", "yielded"], + ) + .map(str::to_owned) + }); + let reported_work_disposition = if run_terminal_state == "succeeded" { + result_disposition + .or(proposed_disposition) + .unwrap_or_else(|| "done".to_owned()) + } else { + result_disposition + .filter(|value| value != "done") + .or_else(|| proposed_disposition.filter(|value| value != "done")) + .unwrap_or_else(|| "yielded".to_owned()) + }; + let harness_terminal_proposal = proposal.unwrap_or(Value::Null); + + json!({ + "schema": "paperclip.prp.terminal.v1", + "turnTerminalState": turn_terminal_state, + "runTerminalState": run_terminal_state, + "reportedWorkDisposition": reported_work_disposition, + "harnessTerminalProposal": harness_terminal_proposal, + "reconciliation": { + "authority": "runner", + "reason": reason, + "proposalValid": proposal_valid, + "proposalContradicted": proposal_contradicted, + "processExit": exit, + "semanticResultObserved": semantic_result.is_some(), + } + }) +} + struct RunnerState { config: RunnerConfig, source_seq: u64, @@ -462,6 +636,7 @@ struct RunnerState { logs: BoundedLogBuffer, semantic_result: Option, pending_terminal: Option, + terminal_intent: TerminalIntent, terminal_emitted: bool, stdout_closed: bool, } @@ -477,6 +652,7 @@ impl RunnerState { harness: None, semantic_result: None, pending_terminal: None, + terminal_intent: TerminalIntent::Natural, terminal_emitted: false, stdout_closed: false, } @@ -584,6 +760,7 @@ impl RunnerState { &self.config.fake_harness_path, &args, self.config.shutdown_grace, + self.config.harness_max_line_bytes, )?; let ready_deadline = Instant::now() + Duration::from_secs(2); loop { @@ -612,6 +789,9 @@ impl RunnerState { )); } Ok(ProcessOutput::Stderr(line)) => self.logs.push(format!("stderr: {line}")), + Ok(ProcessOutput::StdoutError(message)) => { + return Err(Phase2Error::invalid(message)); + } Ok(ProcessOutput::StdoutClosed) => { return Err(Phase2Error::invalid("fake harness exited before ready")); } @@ -676,7 +856,15 @@ impl RunnerState { )?; self.send_harness_command(&command)?; } - "request.resolve" | "turn.interrupt" | "turn.stop" | "session.close" | "run.cancel" => { + "request.resolve" | "session.close" => { + self.send_harness_command(&command)?; + } + "turn.interrupt" | "turn.stop" => { + self.terminal_intent = TerminalIntent::Interrupted; + self.send_harness_command(&command)?; + } + "run.cancel" => { + self.terminal_intent = TerminalIntent::Cancelled; self.send_harness_command(&command)?; } unsupported => { @@ -717,6 +905,12 @@ impl RunnerState { match output { ProcessOutput::Stdout(line) => self.handle_harness_line(&line)?, ProcessOutput::Stderr(line) => self.logs.push(format!("stderr: {line}")), + ProcessOutput::StdoutError(message) => { + self.logs.push(format!("stdout: [{message}]")); + self.terminal_intent = TerminalIntent::ProtocolViolation; + self.terminate_harness()?; + return Ok(()); + } ProcessOutput::StdoutClosed => self.stdout_closed = true, ProcessOutput::StderrClosed => {} } @@ -854,21 +1048,19 @@ impl RunnerState { self.emit_event( "harness.exited", json!({ - "processExit": exit, + "processExit": &exit, "semanticResultObserved": self.semantic_result.is_some(), "logs": self.logs.snapshot(), }), None, None, )?; - let terminal = self.pending_terminal.take().unwrap_or_else(|| { - json!({ - "schema": "paperclip.prp.terminal.v1", - "turnTerminalState": "failed", - "runTerminalState": "failed", - "reportedWorkDisposition": "yielded" - }) - }); + let terminal = reconcile_terminal( + self.pending_terminal.take(), + &exit, + self.semantic_result.as_ref(), + self.terminal_intent, + ); self.emit_event("run.terminal", terminal, None, None) } } @@ -904,11 +1096,15 @@ pub fn run_local_runner(config: RunnerConfig) -> Result<(), Phase2Error> { if state.harness.is_none() { return Err(Phase2Error::invalid("controller closed before run.prepare")); } + state.terminal_intent = TerminalIntent::ControllerClosed; state.terminate_harness()?; return Ok(()); } } state.drain_harness_output()?; + if state.terminal_emitted { + return Ok(()); + } if state.stdout_closed { state.finish_harness()?; return Ok(()); @@ -1209,6 +1405,76 @@ mod tests { assert!(logs.snapshot().retained_bytes <= 12); } + #[test] + fn bounded_line_reader_rejects_an_oversized_frame_and_recovers() { + let mut source = vec![b'x'; 4_096]; + source.extend_from_slice(b"\nnext\n"); + let mut reader = BufReader::new(std::io::Cursor::new(source)); + + assert_eq!( + read_bounded_line(&mut reader, 64).expect("oversized line should be classified"), + BoundedLine::TooLong + ); + assert_eq!( + read_bounded_line(&mut reader, 64).expect("reader should continue after the frame"), + BoundedLine::Line("next".to_owned()) + ); + } + + #[test] + fn nonzero_exit_overrides_a_success_terminal_proposal() { + let terminal = reconcile_terminal( + Some(json!({ + "schema": "paperclip.prp.terminal.v1", + "turnTerminalState": "completed", + "runTerminalState": "succeeded", + "reportedWorkDisposition": "done" + })), + &ProcessExitFact { + exit_code: Some(7), + success: false, + signal: None, + }, + Some(&json!({ "reportedWorkDisposition": "done" })), + TerminalIntent::Natural, + ); + + assert_eq!(terminal["turnTerminalState"], "failed"); + assert_eq!(terminal["runTerminalState"], "failed"); + assert_eq!(terminal["reportedWorkDisposition"], "yielded"); + assert_eq!(terminal["reconciliation"]["authority"], "runner"); + assert_eq!(terminal["reconciliation"]["proposalValid"], true); + assert_eq!(terminal["reconciliation"]["proposalContradicted"], true); + assert_eq!( + terminal["harnessTerminalProposal"]["runTerminalState"], + "succeeded" + ); + } + + #[test] + fn successful_process_and_result_override_a_failed_terminal_proposal() { + let terminal = reconcile_terminal( + Some(json!({ + "schema": "paperclip.prp.terminal.v1", + "turnTerminalState": "failed", + "runTerminalState": "failed", + "reportedWorkDisposition": "yielded" + })), + &ProcessExitFact { + exit_code: Some(0), + success: true, + signal: None, + }, + Some(&json!({ "reportedWorkDisposition": "done" })), + TerminalIntent::Natural, + ); + + assert_eq!(terminal["turnTerminalState"], "completed"); + assert_eq!(terminal["runTerminalState"], "succeeded"); + assert_eq!(terminal["reportedWorkDisposition"], "done"); + assert_eq!(terminal["reconciliation"]["proposalContradicted"], true); + } + #[test] fn canonical_json_sorts_object_keys_without_reordering_arrays() { assert_eq!( @@ -1228,6 +1494,7 @@ mod tests { "error", "duplicate-terminal", "linger", + "oversized-line", ] { let script = load_fake_harness_script(&scripts.join(format!("{name}.json"))) .expect("Phase 2 script should load"); diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/phase2_supervisor.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/phase2_supervisor.rs index 2b016953190..5d2110020de 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/phase2_supervisor.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/phase2_supervisor.rs @@ -29,6 +29,7 @@ fn process_group_cleanup_stops_harness_and_worker() { "1".to_owned(), ], Duration::from_millis(50), + 64 * 1024, ) .expect("fake harness should start"); let harness_pid = process.id(); @@ -80,3 +81,68 @@ fn process_group_cleanup_stops_harness_and_worker() { assert!(!process_exists(u64::from(harness_pid))); assert!(!process_exists(worker_pid)); } + +#[test] +fn oversized_harness_stdout_frame_is_rejected() { + let harness = PathBuf::from(env!("CARGO_BIN_EXE_fake-harness")); + let script = PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .join("../../../protocol/fixtures/phase-02/scripts/oversized-line.json"); + let mut process = SupervisedProcess::spawn( + &harness, + &[ + "--script".to_owned(), + script.display().to_string(), + "--delay-ms".to_owned(), + "1".to_owned(), + ], + Duration::from_millis(50), + 512, + ) + .expect("fake harness should start"); + process + .receive_stdout_line(Duration::from_secs(1)) + .expect("ready frame should fit") + .expect("ready frame should exist"); + process + .send(&HarnessCommand { + schema: "paperclip.fake_harness.command.v1".to_owned(), + command_id: "open".to_owned(), + command_type: "session.open".to_owned(), + payload: json!({}), + }) + .expect("session.open should send"); + process + .receive_stdout_line(Duration::from_secs(1)) + .expect("session frame should fit") + .expect("session frame should exist"); + process + .send(&HarnessCommand { + schema: "paperclip.fake_harness.command.v1".to_owned(), + command_id: "turn".to_owned(), + command_type: "turn.start".to_owned(), + payload: json!({ "turnId": "turn_oversized" }), + }) + .expect("turn.start should send"); + process + .receive_stdout_line(Duration::from_secs(1)) + .expect("turn frame should fit") + .expect("turn frame should exist"); + + let mut oversized_error = None; + for _ in 0..3 { + match process.receive_stdout_line(Duration::from_secs(1)) { + Ok(Some(_)) => {} + Ok(None) => break, + Err(error) => { + oversized_error = Some(error); + break; + } + } + } + let error = oversized_error.expect("oversized harness frame must be rejected"); + assert!(error.to_string().contains("exceeded 512 bytes")); + + process + .terminate_group() + .expect("oversized-frame harness should be cleaned up"); +} diff --git a/packages/paperclip-runner/scripts/phase2-browser-server.mjs b/packages/paperclip-runner/scripts/phase2-browser-server.mjs index 338007f0545..c523b1c64f1 100644 --- a/packages/paperclip-runner/scripts/phase2-browser-server.mjs +++ b/packages/paperclip-runner/scripts/phase2-browser-server.mjs @@ -1,23 +1,214 @@ import { randomUUID } from "node:crypto"; -function readJson(request) { +export const PHASE2_BROWSER_LIMITS = Object.freeze({ + maxActiveRuns: 4, + maxRetainedRuns: 16, + maxRequestBodyBytes: 16 * 1024, + maxStreamClientsPerRun: 8, +}); + +class Phase2HttpError extends Error { + constructor(status, message) { + super(message); + this.status = status; + } +} + +function positiveInteger(value, name) { + if (!Number.isSafeInteger(value) || value < 1) { + throw new TypeError(`${name} must be a positive safe integer`); + } + return value; +} + +function normalizeLimits(overrides = {}) { + const limits = { + maxActiveRuns: positiveInteger( + overrides.maxActiveRuns ?? PHASE2_BROWSER_LIMITS.maxActiveRuns, + "maxActiveRuns", + ), + maxRetainedRuns: positiveInteger( + overrides.maxRetainedRuns ?? PHASE2_BROWSER_LIMITS.maxRetainedRuns, + "maxRetainedRuns", + ), + maxRequestBodyBytes: positiveInteger( + overrides.maxRequestBodyBytes ?? PHASE2_BROWSER_LIMITS.maxRequestBodyBytes, + "maxRequestBodyBytes", + ), + maxStreamClientsPerRun: positiveInteger( + overrides.maxStreamClientsPerRun ?? + PHASE2_BROWSER_LIMITS.maxStreamClientsPerRun, + "maxStreamClientsPerRun", + ), + }; + if (limits.maxRetainedRuns < limits.maxActiveRuns) { + throw new TypeError("maxRetainedRuns must be at least maxActiveRuns"); + } + return limits; +} + +function isLoopbackAddress(address) { + if (typeof address !== "string") { + return false; + } + const normalized = address.toLowerCase().replace(/^::ffff:/, ""); + if (normalized === "::1" || normalized === "[::1]" || normalized === "localhost") { + return true; + } + const octets = normalized.split("."); + return ( + octets.length === 4 && + octets[0] === "127" && + octets.every((octet) => /^\d{1,3}$/.test(octet) && Number(octet) <= 255) + ); +} + +function requestOrigin(request) { + const host = request.headers.host; + if (typeof host !== "string" || host.length === 0) { + throw new Phase2HttpError(403, "Phase 2 requests require a loopback Host"); + } + const protocol = request.socket.encrypted === true ? "https:" : "http:"; + let hostUrl; + try { + hostUrl = new URL(`${protocol}//${host}`); + } catch { + throw new Phase2HttpError(403, "Phase 2 requests require a valid loopback Host"); + } + if (!isLoopbackAddress(hostUrl.hostname)) { + throw new Phase2HttpError(403, "Phase 2 requests require a loopback Host"); + } + const defaultPort = protocol === "https:" ? 443 : 80; + const hostPort = hostUrl.port.length === 0 ? defaultPort : Number(hostUrl.port); + if (hostPort !== request.socket.localPort) { + throw new Phase2HttpError(403, "Phase 2 Host must name the listening loopback port"); + } + return hostUrl.origin; +} + +function assertTrustedTransport(request) { + if ( + !isLoopbackAddress(request.socket.localAddress) || + !isLoopbackAddress(request.socket.remoteAddress) + ) { + throw new Phase2HttpError(403, "Phase 2 is available only over loopback"); + } + const expectedOrigin = requestOrigin(request); + const origin = request.headers.origin; + if (Array.isArray(origin) || (typeof origin === "string" && origin !== expectedOrigin)) { + throw new Phase2HttpError(403, "Phase 2 requests must be same-origin"); + } + const fetchSite = request.headers["sec-fetch-site"]; + if ( + Array.isArray(fetchSite) || + (typeof fetchSite === "string" && !["same-origin", "none"].includes(fetchSite)) + ) { + throw new Phase2HttpError(403, "Phase 2 cross-site requests are forbidden"); + } +} + +function declaredBodyBytes(request) { + const header = request.headers["content-length"]; + if (header === undefined) { + return null; + } + if (Array.isArray(header) || !/^\d+$/.test(header)) { + throw new Phase2HttpError(400, "Invalid Content-Length"); + } + const value = Number(header); + if (!Number.isSafeInteger(value)) { + throw new Phase2HttpError(413, "Phase 2 request body is too large"); + } + return value; +} + +function rejectDeclaredOversize(request, maxBytes) { + const contentLength = declaredBodyBytes(request); + if (contentLength !== null && contentLength > maxBytes) { + request.resume(); + throw new Phase2HttpError( + 413, + `Phase 2 request body exceeds ${maxBytes} bytes`, + ); + } +} + +function readJson(request, maxBytes) { + rejectDeclaredOversize(request, maxBytes); + const contentType = request.headers["content-type"]; + if ( + typeof contentType !== "string" || + !/^application\/json(?:\s*;|$)/i.test(contentType) + ) { + request.resume(); + throw new Phase2HttpError(415, "Phase 2 request bodies must use application/json"); + } + return new Promise((resolve, reject) => { const chunks = []; - request.on("data", (chunk) => chunks.push(chunk)); - request.on("end", () => { - try { - resolve(chunks.length === 0 ? {} : JSON.parse(Buffer.concat(chunks).toString("utf8"))); - } catch (error) { - reject(error); + let receivedBytes = 0; + let settled = false; + + const cleanup = () => { + request.off("data", onData); + request.off("end", onEnd); + request.off("error", onError); + request.off("aborted", onAborted); + }; + const fail = (error, drain = false) => { + if (settled) { + return; } - }); - request.on("error", reject); + settled = true; + chunks.length = 0; + cleanup(); + if (drain) { + request.resume(); + } + reject(error); + }; + const onData = (chunk) => { + const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk); + receivedBytes += buffer.length; + if (receivedBytes > maxBytes) { + fail( + new Phase2HttpError( + 413, + `Phase 2 request body exceeds ${maxBytes} bytes`, + ), + true, + ); + return; + } + chunks.push(buffer); + }; + const onEnd = () => { + if (settled) { + return; + } + settled = true; + cleanup(); + try { + const source = Buffer.concat(chunks, receivedBytes).toString("utf8"); + resolve(source.length === 0 ? {} : JSON.parse(source)); + } catch { + reject(new Phase2HttpError(400, "Phase 2 request body must be valid JSON")); + } + }; + const onError = (error) => fail(error); + const onAborted = () => fail(new Phase2HttpError(400, "Request body was aborted")); + + request.on("data", onData); + request.on("end", onEnd); + request.on("error", onError); + request.on("aborted", onAborted); }); } function json(response, status, value) { response.statusCode = status; response.setHeader("content-type", "application/json; charset=utf-8"); + response.setHeader("cache-control", "no-store"); response.end(`${JSON.stringify(value)}\n`); } @@ -25,24 +216,70 @@ function writeRecord(response, value) { response.write(`${JSON.stringify(value)}\n`); } -export function phase2BrowserServerPlugin() { - const runs = new Map(); +function sendError(response, error, fallbackStatus = 400) { + const status = error instanceof Phase2HttpError ? error.status : fallbackStatus; + const message = error instanceof Error ? error.message : String(error); + json(response, status, { error: message }); +} - async function middleware(request, response, next) { +async function loadPhase2Runner() { + const runnerModuleUrl = new URL( + "../dist/mock-core/phase2-local-runner.js", + import.meta.url, + ).href; + return import(runnerModuleUrl); +} + +export function createPhase2BrowserMiddleware(options = {}) { + const limits = normalizeLimits(options.limits); + const loadRunner = options.loadRunner ?? loadPhase2Runner; + const runs = new Map(); + let startingRuns = 0; + + const activeRunCount = () => + [...runs.values()].filter((entry) => !entry.finished).length; + + const pruneFinishedRuns = (targetSize) => { + for (const [id, entry] of runs) { + if (runs.size <= targetSize) { + return; + } + if (entry.finished) { + runs.delete(id); + } + } + }; + + return async function middleware(request, response, next) { const url = new URL(request.url ?? "/", "http://phase2.local"); if (!url.pathname.startsWith("/api/phase2/")) { next(); return; } - const runnerModuleUrl = new URL( - "../dist/mock-core/phase2-local-runner.js", - import.meta.url, - ).href; - const runner = await import(runnerModuleUrl); + try { + assertTrustedTransport(request); + rejectDeclaredOversize(request, limits.maxRequestBodyBytes); + } catch (error) { + sendError(response, error, 403); + return; + } + if (request.method === "POST" && url.pathname === "/api/phase2/runs") { + if (activeRunCount() + startingRuns >= limits.maxActiveRuns) { + json(response, 429, { error: "Phase 2 active-run limit reached" }); + return; + } + pruneFinishedRuns(limits.maxRetainedRuns - 1); + if (runs.size + startingRuns >= limits.maxRetainedRuns) { + json(response, 429, { error: "Phase 2 retained-run limit reached" }); + return; + } + + startingRuns += 1; try { - const body = await readJson(request); + const body = await readJson(request, limits.maxRequestBodyBytes); + const runner = await loadRunner(); const id = randomUUID(); const history = []; const clients = new Set(); @@ -73,6 +310,7 @@ export function phase2BrowserServerPlugin() { client.end(); } clients.clear(); + pruneFinishedRuns(limits.maxRetainedRuns); }, (error) => { entry.finished = true; @@ -81,16 +319,21 @@ export function phase2BrowserServerPlugin() { client.end(); } clients.clear(); + pruneFinishedRuns(limits.maxRetainedRuns); }, ); json(response, 201, { id, metadata: handle.metadata }); } catch (error) { - json(response, 400, { error: String(error) }); + sendError(response, error); + } finally { + startingRuns -= 1; } return; } - const match = url.pathname.match(/^\/api\/phase2\/runs\/([^/]+)(?:\/(events|interrupt|resolve))?$/); + const match = url.pathname.match( + /^\/api\/phase2\/runs\/([^/]+)(?:\/(events|interrupt|resolve))?$/, + ); if (match === null) { json(response, 404, { error: "Phase 2 route not found" }); return; @@ -102,6 +345,10 @@ export function phase2BrowserServerPlugin() { } const action = match[2]; if (request.method === "GET" && action === "events") { + if (!entry.finished && entry.clients.size >= limits.maxStreamClientsPerRun) { + json(response, 429, { error: "Phase 2 stream-client limit reached" }); + return; + } response.statusCode = 200; response.setHeader("content-type", "application/x-ndjson; charset=utf-8"); response.setHeader("cache-control", "no-store"); @@ -118,26 +365,33 @@ export function phase2BrowserServerPlugin() { } if (request.method === "POST" && action === "interrupt") { try { + await readJson(request, limits.maxRequestBodyBytes); const receipt = await entry.handle.interrupt("browser_operator"); json(response, 200, receipt); } catch (error) { - json(response, 409, { error: String(error) }); + sendError(response, error, 409); } return; } if (request.method === "POST" && action === "resolve") { try { - const body = await readJson(request); - const receipt = await entry.handle.resolveRequest(body.requestId, body.response ?? {}); + const body = await readJson(request, limits.maxRequestBodyBytes); + const receipt = await entry.handle.resolveRequest( + body.requestId, + body.response ?? {}, + ); json(response, 200, receipt); } catch (error) { - json(response, 409, { error: String(error) }); + sendError(response, error, 409); } return; } json(response, 405, { error: "Method not allowed" }); - } + }; +} +export function phase2BrowserServerPlugin(options = {}) { + const middleware = createPhase2BrowserMiddleware(options); return { name: "paperclip-runner-phase2-live-server", configureServer(server) { @@ -148,3 +402,9 @@ export function phase2BrowserServerPlugin() { }, }; } + +export const phase2BrowserServerInternals = { + assertTrustedTransport, + isLoopbackAddress, + readJson, +}; diff --git a/packages/paperclip-runner/scripts/phase2-browser-server.test.mjs b/packages/paperclip-runner/scripts/phase2-browser-server.test.mjs new file mode 100644 index 00000000000..38da92fa5bd --- /dev/null +++ b/packages/paperclip-runner/scripts/phase2-browser-server.test.mjs @@ -0,0 +1,244 @@ +import { createServer, request as httpRequest } from "node:http"; + +import { afterEach, describe, expect, it } from "vitest"; + +import { createPhase2BrowserMiddleware } from "./phase2-browser-server.mjs"; + +const openServers = new Set(); + +function deferred() { + let resolve; + let reject; + const promise = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise; + reject = rejectPromise; + }); + return { promise, reject, resolve }; +} + +function fakeRunner() { + const completions = []; + let startCalls = 0; + return { + completions, + get startCalls() { + return startCalls; + }, + async load() { + return { + async startPhase2Scenario({ scenario }) { + startCalls += 1; + const completion = deferred(); + completions.push(completion); + return { + metadata: { schema: "test", scenario }, + completion: completion.promise, + async interrupt() { + return { status: "accepted" }; + }, + async resolveRequest() { + return { status: "accepted" }; + }, + }; + }, + }; + }, + }; +} + +async function startServer(options = {}) { + const middleware = createPhase2BrowserMiddleware(options); + const server = createServer((request, response) => { + void middleware(request, response, () => { + response.statusCode = 404; + response.end(); + }); + }); + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(0, "127.0.0.1", resolve); + }); + openServers.add(server); + const address = server.address(); + if (typeof address !== "object" || address === null) { + throw new Error("test server did not expose a TCP address"); + } + return { server, port: address.port }; +} + +function sendRequest(port, options = {}) { + const body = options.body ?? ""; + const host = options.host ?? `127.0.0.1:${port}`; + const origin = options.origin === undefined ? `http://${host}` : options.origin; + const headers = { + host, + ...(origin === null ? {} : { origin }), + ...(body.length === 0 ? {} : { "content-type": "application/json" }), + ...options.headers, + }; + return new Promise((resolve, reject) => { + const request = httpRequest( + { + hostname: "127.0.0.1", + port, + path: options.path ?? "/api/phase2/runs", + method: options.method ?? "POST", + headers, + }, + (response) => { + const chunks = []; + response.on("data", (chunk) => chunks.push(chunk)); + response.on("end", () => { + const source = Buffer.concat(chunks).toString("utf8"); + resolve({ + status: response.statusCode, + body: source.length === 0 ? null : JSON.parse(source), + }); + }); + }, + ); + request.on("error", reject); + if (options.chunks !== undefined) { + for (const chunk of options.chunks) { + request.write(chunk); + } + request.end(); + } else { + request.end(body); + } + }); +} + +async function flushPromises() { + await new Promise((resolve) => setImmediate(resolve)); +} + +afterEach(async () => { + const servers = [...openServers]; + openServers.clear(); + await Promise.all( + servers.map( + (server) => + new Promise((resolve) => { + server.close(resolve); + server.closeAllConnections?.(); + }), + ), + ); +}); + +describe("Phase 2 browser transport hardening", () => { + it("denies cross-origin starts and applies the guard before route lookup", async () => { + const runner = fakeRunner(); + const { port } = await startServer({ loadRunner: () => runner.load() }); + + const start = await sendRequest(port, { + origin: "https://attacker.example", + body: JSON.stringify({ scenario: "happy-path" }), + }); + const unknown = await sendRequest(port, { + method: "GET", + path: "/api/phase2/runs/missing/events", + origin: "https://attacker.example", + }); + + expect(start.status).toBe(403); + expect(unknown.status).toBe(403); + expect(runner.startCalls).toBe(0); + }); + + it("rejects non-loopback Host values even when Origin matches", async () => { + const runner = fakeRunner(); + const { port } = await startServer({ loadRunner: () => runner.load() }); + + const response = await sendRequest(port, { + host: "attacker.example", + origin: "http://attacker.example", + body: JSON.stringify({ scenario: "happy-path" }), + }); + + expect(response.status).toBe(403); + expect(runner.startCalls).toBe(0); + }); + + it("rejects a loopback Host that does not name the listening port", async () => { + const runner = fakeRunner(); + const { port } = await startServer({ loadRunner: () => runner.load() }); + + const response = await sendRequest(port, { + host: "127.0.0.1:9", + origin: "http://127.0.0.1:9", + body: JSON.stringify({ scenario: "happy-path" }), + }); + + expect(response.status).toBe(403); + expect(runner.startCalls).toBe(0); + }); + + it("caps active runs and evicts the oldest completed run at the retention bound", async () => { + const runner = fakeRunner(); + const { port } = await startServer({ + loadRunner: () => runner.load(), + limits: { maxActiveRuns: 1, maxRetainedRuns: 2 }, + }); + const body = JSON.stringify({ scenario: "happy-path" }); + + const first = await sendRequest(port, { body }); + const rejected = await sendRequest(port, { body }); + expect(first.status).toBe(201); + expect(rejected.status).toBe(429); + + runner.completions[0].resolve({ run: 1 }); + await flushPromises(); + const second = await sendRequest(port, { body }); + expect(second.status).toBe(201); + runner.completions[1].resolve({ run: 2 }); + await flushPromises(); + + const third = await sendRequest(port, { body }); + expect(third.status).toBe(201); + const evicted = await sendRequest(port, { + method: "GET", + path: `/api/phase2/runs/${first.body.id}/events`, + }); + expect(evicted.status).toBe(404); + + runner.completions[2].resolve({ run: 3 }); + await flushPromises(); + }); + + it("rejects declared and streamed oversized JSON before starting a runner", async () => { + const runner = fakeRunner(); + const { port } = await startServer({ + loadRunner: () => runner.load(), + limits: { maxRequestBodyBytes: 32 }, + }); + const oversized = JSON.stringify({ scenario: "x".repeat(64) }); + + const declared = await sendRequest(port, { body: oversized }); + const streamed = await sendRequest(port, { + chunks: ["{\"scenario\":\"", "y".repeat(64), "\"}"], + headers: { + "content-type": "application/json", + "transfer-encoding": "chunked", + }, + }); + + expect(declared.status).toBe(413); + expect(streamed.status).toBe(413); + expect(runner.startCalls).toBe(0); + }); + + it("requires JSON content so a simple text/plain browser POST cannot start work", async () => { + const runner = fakeRunner(); + const { port } = await startServer({ loadRunner: () => runner.load() }); + + const response = await sendRequest(port, { + body: JSON.stringify({ scenario: "happy-path" }), + headers: { "content-type": "text/plain" }, + }); + + expect(response.status).toBe(415); + expect(runner.startCalls).toBe(0); + }); +}); diff --git a/packages/paperclip-runner/vite.config.ts b/packages/paperclip-runner/vite.config.ts index 121347b763f..3ae16106c1e 100644 --- a/packages/paperclip-runner/vite.config.ts +++ b/packages/paperclip-runner/vite.config.ts @@ -11,6 +11,12 @@ const packageRoot = fileURLToPath(new URL(".", import.meta.url)); export default defineConfig({ root: resolve(packageRoot, "devtools/browser"), plugins: [react(), phase2BrowserServerPlugin()], + server: { + host: "127.0.0.1", + }, + preview: { + host: "127.0.0.1", + }, build: { outDir: resolve(packageRoot, "dist-browser"), emptyOutDir: true, diff --git a/packages/paperclip-runner/vitest.config.ts b/packages/paperclip-runner/vitest.config.ts index c1433e6ef38..2180e29a45a 100644 --- a/packages/paperclip-runner/vitest.config.ts +++ b/packages/paperclip-runner/vitest.config.ts @@ -3,6 +3,6 @@ import { defineConfig } from "vitest/config"; export default defineConfig({ test: { environment: "node", - include: ["src/**/*.test.ts"], + include: ["src/**/*.test.ts", "scripts/phase2-browser-server.test.mjs"], }, });