From 2f28d8112970960dbb9b6b582a3c6cd259337d21 Mon Sep 17 00:00:00 2001 From: nsfisis Date: Mon, 3 Aug 2026 01:05:57 +0900 Subject: feat(php-rpc): rework the RPC channel into the tagged plugin protocol Replace the name\0arg framing with the plugin wire protocol: tagged frames with corr_id multiplexing, a MAX_FRAME_LEN bound, a thread-ID based reentrant SessionLock, and the PluginValue codec (encoder plus the first recursive decoder, iterative with a 512-level depth cap). Float formatting is ported from php-src into shirabe-php-src so the encoder is byte-compatible with serialize() under serialize_precision=-1, which the spawned worker now pins. The worker gains a standing dispatch loop, CallRustMethod reentrancy, hand-written Event proxy stubs, and explicit-error answers for everything not implemented yet. The public query API (get_php_version and friends) is unchanged and now rides the new protocol; the codec is verified against real PHP by roundtrip oracle tests covering floats, non-UTF-8 bytes and deep nesting. Co-Authored-By: Claude Fable 5 --- crates/shirabe-php-rpc/src/frame.rs | 462 ++++++++++++++++++++++++++++++++++++ 1 file changed, 462 insertions(+) create mode 100644 crates/shirabe-php-rpc/src/frame.rs (limited to 'crates/shirabe-php-rpc/src/frame.rs') diff --git a/crates/shirabe-php-rpc/src/frame.rs b/crates/shirabe-php-rpc/src/frame.rs new file mode 100644 index 00000000..3a355c86 --- /dev/null +++ b/crates/shirabe-php-rpc/src/frame.rs @@ -0,0 +1,462 @@ +//! Wire framing: `[u64 length LE][u8 tag][u64 corr_id LE][payload]`, where `length` counts +//! everything after itself (tag + corr_id + payload) and the payload is the frame's remaining +//! fields as a PHP-serialized list. See `docs/dev/php-rpc.md`. + +use crate::value::{self, PluginValue}; +use indexmap::IndexMap; +use std::io::{Read as _, Write as _}; +use std::os::unix::net::UnixStream; + +/// Upper bound for the declared frame length, so a corrupted length header cannot make the +/// process allocate absurd amounts of memory. +pub const MAX_FRAME_LEN: u64 = 256 * 1024 * 1024; + +pub const TAG_CALL_FUNCTION: u8 = 0x00; +pub const TAG_CALL_STATIC_METHOD: u8 = 0x01; +pub const TAG_NEW_OBJECT: u8 = 0x02; +pub const TAG_CALL_PHP_METHOD: u8 = 0x03; +pub const TAG_CALL_RUST_METHOD: u8 = 0x04; +pub const TAG_RETURN: u8 = 0x05; +pub const TAG_THROW: u8 = 0x06; +pub const TAG_RELEASE_RUST_HANDLE: u8 = 0x07; +pub const TAG_RELEASE_PHP_HANDLE: u8 = 0x08; +pub const TAG_EPOCH_BUMP: u8 = 0x09; + +#[derive(Debug)] +pub enum Frame { + CallFunction { + corr_id: u64, + function_name: String, + args: Vec, + out_param_positions: Vec, + }, + CallStaticMethod { + corr_id: u64, + pclass: String, + method_name: String, + args: Vec, + out_param_positions: Vec, + }, + NewObject { + corr_id: u64, + pclass: String, + ctor_args: Vec, + }, + CallPhpMethod { + corr_id: u64, + phandle: u64, + method_name: String, + args: Vec, + out_param_positions: Vec, + }, + CallRustMethod { + corr_id: u64, + rhandle: u64, + method_name: String, + args: Vec, + out_param_positions: Vec, + }, + Return { + corr_id: u64, + value: PluginValue, + out_params: IndexMap, + }, + Throw { + corr_id: u64, + exception_class: String, + message: String, + code: i64, + }, + ReleaseRustHandle { + rhandle: u64, + }, + ReleasePhpHandle { + phandle: u64, + }, + EpochBump { + rhandle: u64, + epoch: u64, + }, +} + +impl Frame { + fn tag(&self) -> u8 { + match self { + Frame::CallFunction { .. } => TAG_CALL_FUNCTION, + Frame::CallStaticMethod { .. } => TAG_CALL_STATIC_METHOD, + Frame::NewObject { .. } => TAG_NEW_OBJECT, + Frame::CallPhpMethod { .. } => TAG_CALL_PHP_METHOD, + Frame::CallRustMethod { .. } => TAG_CALL_RUST_METHOD, + Frame::Return { .. } => TAG_RETURN, + Frame::Throw { .. } => TAG_THROW, + Frame::ReleaseRustHandle { .. } => TAG_RELEASE_RUST_HANDLE, + Frame::ReleasePhpHandle { .. } => TAG_RELEASE_PHP_HANDLE, + Frame::EpochBump { .. } => TAG_EPOCH_BUMP, + } + } + + /// One-way notifications carry no correlation id; the field is 0 on the wire. + fn corr_id(&self) -> u64 { + match self { + Frame::CallFunction { corr_id, .. } + | Frame::CallStaticMethod { corr_id, .. } + | Frame::NewObject { corr_id, .. } + | Frame::CallPhpMethod { corr_id, .. } + | Frame::CallRustMethod { corr_id, .. } + | Frame::Return { corr_id, .. } + | Frame::Throw { corr_id, .. } => *corr_id, + Frame::ReleaseRustHandle { .. } + | Frame::ReleasePhpHandle { .. } + | Frame::EpochBump { .. } => 0, + } + } + + fn fields(&self) -> Vec { + match self { + Frame::CallFunction { + function_name, + args, + out_param_positions, + .. + } => vec![ + PluginValue::string(function_name.clone()), + PluginValue::List(args.clone()), + positions_value(out_param_positions), + ], + Frame::CallStaticMethod { + pclass, + method_name, + args, + out_param_positions, + .. + } => vec![ + PluginValue::string(pclass.clone()), + PluginValue::string(method_name.clone()), + PluginValue::List(args.clone()), + positions_value(out_param_positions), + ], + Frame::NewObject { + pclass, ctor_args, .. + } => vec![ + PluginValue::string(pclass.clone()), + PluginValue::List(ctor_args.clone()), + ], + Frame::CallPhpMethod { + phandle, + method_name, + args, + out_param_positions, + .. + } => vec![ + int_value(*phandle), + PluginValue::string(method_name.clone()), + PluginValue::List(args.clone()), + positions_value(out_param_positions), + ], + Frame::CallRustMethod { + rhandle, + method_name, + args, + out_param_positions, + .. + } => vec![ + int_value(*rhandle), + PluginValue::string(method_name.clone()), + PluginValue::List(args.clone()), + positions_value(out_param_positions), + ], + Frame::Return { + value, out_params, .. + } => vec![ + value.clone(), + PluginValue::Array( + out_params + .iter() + .map(|(pos, v)| (pos.to_string().into_bytes(), v.clone())) + .collect(), + ), + ], + Frame::Throw { + exception_class, + message, + code, + .. + } => vec![ + PluginValue::string(exception_class.clone()), + PluginValue::string(message.clone()), + PluginValue::Int(*code), + ], + Frame::ReleaseRustHandle { rhandle } => vec![int_value(*rhandle)], + Frame::ReleasePhpHandle { phandle } => vec![int_value(*phandle)], + Frame::EpochBump { rhandle, epoch } => vec![int_value(*rhandle), int_value(*epoch)], + } + } +} + +fn int_value(id: u64) -> PluginValue { + PluginValue::Int(i64::try_from(id).expect("handle id exceeds i64")) +} + +fn positions_value(positions: &[u32]) -> PluginValue { + PluginValue::List( + positions + .iter() + .map(|p| PluginValue::Int(i64::from(*p))) + .collect(), + ) +} + +pub fn write_frame(stream: &mut UnixStream, frame: &Frame) -> std::io::Result<()> { + let payload = value::serialize(&PluginValue::List(frame.fields())); + let len = 1 + 8 + payload.len() as u64; + stream.write_all(&len.to_le_bytes())?; + stream.write_all(&[frame.tag()])?; + stream.write_all(&frame.corr_id().to_le_bytes())?; + stream.write_all(&payload)?; + stream.flush() +} + +pub fn read_frame(stream: &mut UnixStream) -> std::io::Result { + let mut header = [0u8; 8]; + stream.read_exact(&mut header)?; + let len = u64::from_le_bytes(header); + if len > MAX_FRAME_LEN { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!("frame length {len} exceeds MAX_FRAME_LEN"), + )); + } + if len < 9 { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!("frame length {len} is shorter than the tag and corr_id fields"), + )); + } + let mut tag = [0u8; 1]; + stream.read_exact(&mut tag)?; + let mut corr_id = [0u8; 8]; + stream.read_exact(&mut corr_id)?; + let mut payload = vec![0u8; (len - 9) as usize]; + stream.read_exact(&mut payload)?; + Ok(decode_frame(tag[0], u64::from_le_bytes(corr_id), &payload)) +} + +/// Both sides of this protocol are written in the same commit of Shirabe, so a frame that does +/// not decode is a programming error, not a runtime condition; every mismatch panics. +fn decode_frame(tag: u8, corr_id: u64, payload: &[u8]) -> Frame { + let fields = match value::unserialize(payload) { + Ok(PluginValue::List(fields)) => fields, + other => panic!("PHP RPC: protocol violation — frame payload is not a list: {other:?}"), + }; + let mut fields = fields.into_iter(); + let mut next = || { + fields + .next() + .unwrap_or_else(|| panic!("PHP RPC: protocol violation — missing frame field")) + }; + match tag { + TAG_CALL_FUNCTION => Frame::CallFunction { + corr_id, + function_name: expect_string(next()), + args: expect_list(next()), + out_param_positions: expect_positions(next()), + }, + TAG_CALL_STATIC_METHOD => Frame::CallStaticMethod { + corr_id, + pclass: expect_string(next()), + method_name: expect_string(next()), + args: expect_list(next()), + out_param_positions: expect_positions(next()), + }, + TAG_NEW_OBJECT => Frame::NewObject { + corr_id, + pclass: expect_string(next()), + ctor_args: expect_list(next()), + }, + TAG_CALL_PHP_METHOD => Frame::CallPhpMethod { + corr_id, + phandle: expect_id(next()), + method_name: expect_string(next()), + args: expect_list(next()), + out_param_positions: expect_positions(next()), + }, + TAG_CALL_RUST_METHOD => Frame::CallRustMethod { + corr_id, + rhandle: expect_id(next()), + method_name: expect_string(next()), + args: expect_list(next()), + out_param_positions: expect_positions(next()), + }, + TAG_RETURN => Frame::Return { + corr_id, + value: next(), + out_params: expect_out_params(next()), + }, + TAG_THROW => Frame::Throw { + corr_id, + exception_class: expect_string(next()), + message: expect_string(next()), + code: match next() { + PluginValue::Int(code) => code, + other => { + panic!("PHP RPC: protocol violation — Throw code is not an int: {other:?}") + } + }, + }, + TAG_RELEASE_RUST_HANDLE => Frame::ReleaseRustHandle { + rhandle: expect_id(next()), + }, + TAG_RELEASE_PHP_HANDLE => Frame::ReleasePhpHandle { + phandle: expect_id(next()), + }, + TAG_EPOCH_BUMP => Frame::EpochBump { + rhandle: expect_id(next()), + epoch: expect_id(next()), + }, + _ => panic!("PHP RPC: protocol violation — unknown frame tag {tag:#04x}"), + } +} + +fn expect_string(value: PluginValue) -> String { + match value { + PluginValue::String(bytes) => String::from_utf8(bytes) + .unwrap_or_else(|e| panic!("PHP RPC: protocol violation — non-UTF-8 name field: {e}")), + other => panic!("PHP RPC: protocol violation — expected a string field: {other:?}"), + } +} + +fn expect_list(value: PluginValue) -> Vec { + match value { + PluginValue::List(items) => items, + other => panic!("PHP RPC: protocol violation — expected a list field: {other:?}"), + } +} + +fn expect_id(value: PluginValue) -> u64 { + match value { + PluginValue::Int(n) => u64::try_from(n) + .unwrap_or_else(|_| panic!("PHP RPC: protocol violation — negative handle id {n}")), + other => panic!("PHP RPC: protocol violation — expected an int id field: {other:?}"), + } +} + +fn expect_positions(value: PluginValue) -> Vec { + expect_list(value) + .into_iter() + .map(|item| match item { + PluginValue::Int(n) => u32::try_from(n).unwrap_or_else(|_| { + panic!("PHP RPC: protocol violation — out param position {n} out of range") + }), + other => { + panic!("PHP RPC: protocol violation — out param position is not an int: {other:?}") + } + }) + .collect() +} + +fn expect_out_params(value: PluginValue) -> IndexMap { + match value { + PluginValue::List(items) if items.is_empty() => IndexMap::new(), + PluginValue::List(items) => items + .into_iter() + .enumerate() + .map(|(index, item)| (index as u32, item)) + .collect(), + PluginValue::Array(map) => map + .into_iter() + .map(|(key, item)| { + let position = std::str::from_utf8(&key) + .ok() + .and_then(|s| s.parse().ok()) + .unwrap_or_else(|| { + panic!( + "PHP RPC: protocol violation — out param key is not a position: {:?}", + String::from_utf8_lossy(&key) + ) + }); + (position, item) + }) + .collect(), + other => panic!("PHP RPC: protocol violation — out_params is not an array: {other:?}"), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn roundtrip(frame: Frame) -> Frame { + let (mut a, mut b) = UnixStream::pair().unwrap(); + write_frame(&mut a, &frame).unwrap(); + read_frame(&mut b).unwrap() + } + + #[test] + fn frame_roundtrip_call_function() { + let frame = roundtrip(Frame::CallFunction { + corr_id: 7, + function_name: "defined".to_string(), + args: vec![PluginValue::string("PHP_VERSION")], + out_param_positions: vec![], + }); + match frame { + Frame::CallFunction { + corr_id, + function_name, + args, + out_param_positions, + } => { + assert_eq!(corr_id, 7); + assert_eq!(function_name, "defined"); + assert_eq!(args, vec![PluginValue::string("PHP_VERSION")]); + assert!(out_param_positions.is_empty()); + } + other => panic!("unexpected frame: {other:?}"), + } + } + + #[test] + fn frame_roundtrip_return_with_out_params() { + let frame = roundtrip(Frame::Return { + corr_id: 9, + value: PluginValue::Bool(true), + out_params: [(2u32, PluginValue::string("x"))].into_iter().collect(), + }); + match frame { + Frame::Return { + corr_id, + value, + out_params, + } => { + assert_eq!(corr_id, 9); + assert_eq!(value, PluginValue::Bool(true)); + assert_eq!(out_params.get(&2), Some(&PluginValue::string("x"))); + } + other => panic!("unexpected frame: {other:?}"), + } + } + + #[test] + fn frame_roundtrip_one_way_notification() { + let frame = roundtrip(Frame::EpochBump { + rhandle: 4, + epoch: 2, + }); + match frame { + Frame::EpochBump { rhandle, epoch } => { + assert_eq!(rhandle, 4); + assert_eq!(epoch, 2); + } + other => panic!("unexpected frame: {other:?}"), + } + } + + #[test] + fn read_frame_rejects_oversized_length() { + let (mut a, mut b) = UnixStream::pair().unwrap(); + a.write_all(&(MAX_FRAME_LEN + 1).to_le_bytes()).unwrap(); + let err = read_frame(&mut b).unwrap_err(); + assert_eq!(err.kind(), std::io::ErrorKind::InvalidData); + assert!(err.to_string().contains("MAX_FRAME_LEN"), "{err}"); + } +} -- cgit v1.3.1