fix(runner): harden phase 2 local boundaries

Enforce loopback/same-origin transport and bounded browser resources, frame harness output before allocation, and reconcile terminal state from runner-owned facts.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-08-07 21:32:24 +00:00
1 parent 90daf4ff00
commit bfef34393c
9 files changed
+975 -77

No files matched your search

@@ -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
@@ -0,0 +1,12 @@
{
"schema": "paperclip.fake_harness.script.v1",
"name": "oversized-line",
"steps": [
{
"kind": "log",
"stream": "stdout",
"text": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
},
{ "kind": "exit", "code": 0 }
]
}
@@ -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),
),
@@ -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<u64>,
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<R: BufRead>(reader: &mut R, max_bytes: usize) -> io::Result<BoundedLine> {
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<R: io::Read + Send + 'static>(
reader: R,
sender: mpsc::Sender<ProcessOutput>,
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<Self, Phase2Error> {
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<Value>,
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<Value>,
pending_terminal: Option<Value>,
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");
@@ -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");
}
@@ -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,
};
@@ -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);
});
});
+6
View File
@@ -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,
+1 -1
View File
@@ -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"],
},
});