rsbinder 0.11.0

// Copyright 2022 Jeff Kim <hiking90@gmail.com>
// SPDX-License-Identifier: Apache-2.0

//! Transport abstraction for the RPC stack.
//!
//! A [`RpcTransport`] carries **length-framed byte messages** for one
//! RPC connection and reports the [`PeerIdentity`] of the other end.
//! The implementation *defines the trust boundary*: a `unix` socket
//! trusts filesystem permissions + `SO_PEERCRED`; `vsock` trusts
//! hypervisor VM isolation; `tls` trusts a certificate; the gated
//! `tcp_debug` backend trusts **nothing** and is debug/interop only.
//!
//! Framing is the transport's responsibility (not the wire codec's), so
//! the wire layer can think purely in whole messages. Stream backends
//! (`unix`, `tcp_debug`) share the length-prefix helpers in this
//! module; the in-process `mem` backend frames implicitly (one channel
//! message == one frame).
//!
//! The trait is **synchronous / blocking** (matches android-12 r34's
//! blocking-thread model). An `async` adapter can be layered *on top*
//! without changing this trait.

use std::fmt;
use std::io::{ErrorKind, Read, Write};

use super::{RpcError, RpcResult};

mod mem;
#[cfg(feature = "rpc-tcp-debug")]
mod tcp_debug;
#[cfg(feature = "rpc-tls")]
mod tls;
pub(crate) mod unix;
#[cfg(all(feature = "rpc-vsock", any(target_os = "linux", target_os = "android")))]
mod vsock;

pub use mem::MemTransport;
#[cfg(feature = "rpc-tcp-debug")]
pub use tcp_debug::{insecure_warning_emitted, TcpDebugTransport};
#[cfg(feature = "rpc-tls")]
pub use tls::{TlsStream, TlsTransport};
pub use unix::UnixTransport;
#[cfg(all(feature = "rpc-vsock", any(target_os = "linux", target_os = "android")))]
pub use vsock::VsockTransport;

/// Hard cap on a single decoded frame.
///
/// A length header declaring more than this is rejected **before any
/// allocation** — an adversarial peer cannot trigger an OOM by claiming
/// a huge body. 64 MiB is far above any legitimate binder transaction
/// yet bounded.
pub const MAX_FRAME_LEN: usize = 64 * 1024 * 1024;

/// One RPC connection: framed byte transport + peer identity.
///
/// Synchronous and blocking. `&self` (not `&mut self`) so a session can
/// hold one transport and use it from a sender thread and a receiver
/// thread concurrently — full-duplex sockets and the `mem` channel pair
/// both support that without a deadlock. Implementations must keep
/// `send_frame`/`recv_frame` independently callable from two threads.
pub trait RpcTransport: Send + Sync {
    /// Send exactly one logical frame. The implementation guarantees
    /// framing (length prefix or channel message boundary).
    fn send_frame(&self, buf: &[u8]) -> RpcResult<()>;

    /// Receive exactly one logical frame.
    ///
    /// A clean peer close with nothing pending is
    /// [`RpcError::EndOfStream`]; a header received but body short is
    /// [`RpcError::Truncated`]. Never panics or loops forever on a
    /// hostile peer.
    fn recv_frame(&self) -> RpcResult<Vec<u8>>;

    /// The other end's identity, as established by this transport.
    ///
    /// This is the RPC equivalent of kernel binder's
    /// `getCallingUid()`/SELinux context — **but only as strong as the
    /// transport's trust boundary**. [`PeerIdentity::Anonymous`] means
    /// no identity at all (no ACL possible); callers must log it as
    /// such and not grant trust.
    fn peer_identity(&self) -> PeerIdentity;

    /// Short human-readable description for diagnostics/logging
    /// (e.g. socket path, `"mem"`, vsock cid). Never carries secrets.
    fn describe(&self) -> &str;

    /// Set a read deadline for subsequent [`RpcTransport::recv_frame`]
    /// calls. `None` clears it (fully blocking). The
    /// default is a no-op for backends with no read-timeout notion;
    /// `unix` / `mem` / `tcp_debug` / `vsock` / `tls` override it. A deadline that
    /// elapses with **nothing consumed** surfaces as
    /// [`RpcError::Timeout`] (the stream stays frame-synchronized); a
    /// deadline that elapses mid-frame is [`RpcError::DeadlineMidFrame`].
    fn set_read_timeout(&self, _timeout: Option<std::time::Duration>) -> RpcResult<()> {
        Ok(())
    }

    /// Set a write deadline for subsequent sends. `None` clears it. The
    /// default is a no-op for backends with no write-timeout notion
    /// (`mem`'s send never blocks on a peer); socket-backed transports
    /// (`unix` / `vsock` / `tcp_debug` / `tls`) override it. This bounds
    /// the reply-send phase so a peer that completes the handshake/
    /// admission and then stops reading cannot pin its worker thread (and,
    /// under [`set_max_connections`](super::server::RpcServer::set_max_connections),
    /// its admission slot) forever by stalling our blocking `write_all`
    /// once the kernel send buffer fills.
    fn set_write_timeout(&self, _timeout: Option<std::time::Duration>) -> RpcResult<()> {
        Ok(())
    }

    /// Shut the connection down in both directions: wake a reader blocked
    /// in [`recv_frame`](Self::recv_frame) / [`recv_raw`](Self::recv_raw)
    /// and make this end's later sends fail. (The *peer's* sends are the
    /// platform's business — a Linux `AF_UNIX` peer gets `EPIPE` at once,
    /// a macOS one has its writes accepted and discarded, a TCP peer on
    /// either sees the reset on a later write; it reads the end of stream
    /// either way.)
    /// This is how a session ends its
    /// connections — `RpcSession::close_session`, `RpcServer::terminate`, a
    /// slot retired after a lost stream. Required, with no default on
    /// purpose: a transport that silently did nothing here would leave a
    /// serve loop or an incoming-connection thread parked in `recv`
    /// forever, and `RpcSession::close_session` would hang on the join. The
    /// same reasoning already made `TlsStream::shutdown_stream` required
    /// (the `rpc-tls` backend).
    ///
    /// The contract is narrower than the name suggests, and every caller
    /// in rsbinder is written to it (plan 2-21 §3.4):
    ///
    /// - **What happens to bytes already received is platform- and
    ///   backend-dependent, and nothing may assume it.** A kernel may keep
    ///   its receive queue across a local shutdown — Linux: a reader gets
    ///   the queued bytes, then end of stream — or drop it (macOS). A
    ///   transport's own buffer may survive (`tls`'s decrypted plaintext)
    ///   or be cleared (`unix` fd-mode clears its leftover here). A caller
    ///   that must not read what is buffered keeps that decision in
    ///   session state — the slot's `unreadable` mark — not here. `mem`
    ///   models the Linux behaviour, so a hermetic test exercises the case
    ///   that hides bugs.
    /// - **A reader woken by this returns the end of stream** —
    ///   [`RpcError::EndOfStream`] at a frame boundary, [`RpcError::Truncated`]
    ///   mid-frame — never a distinct "shut down locally" error. Who ended
    ///   the connection is session knowledge (`SessionEnd::by`), not
    ///   transport knowledge; a TLS transport does not report its own
    ///   shutdown as an unclean end.
    /// - **Idempotent.** A second call returns `Ok(())`; the socket
    ///   backends absorb the `ENOTCONN` a second shutdown raises on macOS.
    /// - **The `Err` is diagnostic.** The connection is being ended
    ///   whatever it says; callers log it and never branch on it.
    /// - **This is not `close`.** The kernel's queue is certainly gone only
    ///   when the last `Arc<dyn RpcTransport>` drops.
    ///
    /// The measured behaviour of `unix` (both modes), `mem`, `tcp_debug`
    /// and `tls` on both platforms is pinned by
    /// `tests/rpc_transport_conformance.rs`. `vsock` is not measured: it
    /// needs a VM peer, so its behaviour here is inferred from `vsock(7)`
    /// and covered only by its own `#[ignore]`d tests.
    fn shutdown(&self) -> RpcResult<()>;

    /// Send one frame plus passed file descriptors out-of-band (opt-in
    /// `FileDescriptorTransportMode::Unix`).
    ///
    /// The **default rejects any fd** — so `mem`/`vsock`/`tls` are
    /// fd-incapable *by type*, with no extra code. Only `unix` overrides
    /// this with `SCM_RIGHTS`. An empty `fds` slice falls back to the
    /// plain framed send.
    fn send_frame_with_fds(
        &self,
        buf: &[u8],
        fds: &[std::os::fd::BorrowedFd<'_>],
    ) -> RpcResult<()> {
        if fds.is_empty() {
            self.send_frame(buf)
        } else {
            Err(RpcError::Protocol(
                "this transport cannot pass file descriptors (UDS only)",
            ))
        }
    }

    /// Receive one frame plus any out-of-band file descriptors.
    /// Default: never yields fds (the plain framed recv); only `unix`
    /// overrides with `SCM_RIGHTS`.
    fn recv_frame_with_fds(&self) -> RpcResult<(Vec<u8>, Vec<std::os::fd::OwnedFd>)> {
        Ok((self.recv_frame()?, Vec::new()))
    }

    /// Send raw bytes with **no framing**. The real android RPC wire
    /// has no length prefix (`RpcState::rpcSend` writes the
    /// `RpcWireHeader` + body directly) — the android-13+ profile drives
    /// framing itself via `wire_android13`. The default is
    /// **unsupported**: right for a frame-only backend (`mem`), and a
    /// silent trap for a byte-stream one — an android-13+ session over a
    /// backend that forgot it fails at its first handshake byte (it
    /// happened to `vsock`, then to `tcp_debug`). Every stream backend
    /// (`unix`, `tcp_debug`, `vsock`, `tls`) must override both this and
    /// [`recv_raw`](Self::recv_raw). The existing R34 path never calls
    /// this — `send_frame`/`recv_frame` are byte-unchanged.
    fn send_raw(&self, _buf: &[u8]) -> RpcResult<()> {
        Err(RpcError::Protocol("this transport has no raw byte access"))
    }

    /// Read up to `buf.len()` raw bytes (one `read`; `Ok(0)` = peer
    /// closed). Pairs with [`RpcTransport::send_raw`]. Default:
    /// unsupported (see [`RpcTransport::send_raw`]).
    fn recv_raw(&self, _buf: &mut [u8]) -> RpcResult<usize> {
        Err(RpcError::Protocol("this transport has no raw byte access"))
    }

    /// Send raw bytes with **no framing**, passing `fds` out-of-band via
    /// `SCM_RIGHTS` (the android-13+ v1+ `Unix` FD-over-RPC path). This
    /// is [`RpcTransport::send_raw`] + the ancillary channel of
    /// [`RpcTransport::send_frame_with_fds`],
    /// minus the length prefix: the real android RPC wire has none
    /// (`RpcWireHeader.bodySize` is authoritative) and AOSP rides the
    /// fds on the **first** `sendmsg` of the message
    /// (`RpcTransportRaw::interruptableWriteFully`, `sentFds`). An empty
    /// `fds` slice is exactly [`RpcTransport::send_raw`]. Default:
    /// unsupported unless `fds` is empty (frame-only transports stay
    /// fd-incapable *by type*, no extra code); only `unix` overrides.
    fn send_raw_with_fds(&self, buf: &[u8], fds: &[std::os::fd::BorrowedFd<'_>]) -> RpcResult<()> {
        if fds.is_empty() {
            self.send_raw(buf)
        } else {
            Err(RpcError::Protocol(
                "this transport cannot pass file descriptors (UDS only)",
            ))
        }
    }

    /// Read up to `buf.len()` raw bytes (one `recvmsg`; `Ok((0, _))` =
    /// peer closed) plus any `SCM_RIGHTS` fds delivered with them.
    /// Pairs with
    /// [`RpcTransport::send_raw_with_fds`]; received fds are
    /// `O_CLOEXEC`. AOSP accumulates ancillary fds across the
    /// `recvmsg`s that read one message
    /// (`RpcTransportRaw::interruptableReadFully`), so the caller
    /// gathers fds across the header+body reads. Default: never yields
    /// fds (plain [`RpcTransport::recv_raw`]); only `unix` overrides.
    fn recv_raw_with_fds(&self, buf: &mut [u8]) -> RpcResult<(usize, Vec<std::os::fd::OwnedFd>)> {
        Ok((self.recv_raw(buf)?, Vec::new()))
    }
}

/// Identity of the peer on the other end of a [`RpcTransport`].
///
/// `#[non_exhaustive]`: matching code must keep a wildcard arm, since
/// further variants may be added.
#[derive(Clone, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub enum PeerIdentity {
    /// A local peer whose credentials the kernel vouches for
    /// (`SO_PEERCRED` over a Unix domain socket, or the current
    /// process for the in-memory test transport).
    Local {
        /// Peer process effective UID.
        uid: u32,
        /// Peer process PID (`-1` if unavailable on this platform).
        pid: i32,
    },
    /// A vsock peer, identified by its context id. **Not an ACL
    /// basis** — `cid` is a routing address, and the trust boundary is
    /// hypervisor VM isolation. Logged with the cid.
    Vsock {
        /// vsock context id of the peer VM/host.
        cid: u32,
    },
    /// A TLS peer authenticated by its leaf certificate. The trust
    /// boundary is the certificate chain.
    Certificate(CertId),
    /// No identity is available. **ACL is not possible** against an
    /// anonymous peer; this must be surfaced in logs and never treated
    /// as trusted. Used by the debug-only plaintext TCP backend.
    Anonymous,
}

/// Identity extracted from a peer's TLS leaf certificate. Carries the
/// subject and a SHA-256 fingerprint; ACL is the caller's
/// responsibility on top of this.
#[derive(Clone, PartialEq, Eq)]
pub struct CertId {
    subject: String,
    fingerprint: [u8; 32],
}

impl CertId {
    /// Construct from a subject string and the leaf cert SHA-256.
    pub fn new(subject: impl Into<String>, fingerprint: [u8; 32]) -> Self {
        CertId {
            subject: subject.into(),
            fingerprint,
        }
    }
    /// The certificate subject (DN / SAN summary).
    pub fn subject(&self) -> &str {
        &self.subject
    }
    /// The leaf certificate SHA-256 fingerprint.
    pub fn fingerprint(&self) -> &[u8; 32] {
        &self.fingerprint
    }
    /// Lowercase hex of the fingerprint (for logging / pinning).
    pub fn fingerprint_hex(&self) -> String {
        self.fingerprint
            .iter()
            .map(|b| format!("{b:02x}"))
            .collect()
    }
}

impl fmt::Debug for CertId {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("CertId")
            .field("subject", &self.subject)
            .field("fingerprint", &self.fingerprint_hex())
            .finish()
    }
}

impl PeerIdentity {
    /// `true` for [`PeerIdentity::Local`].
    pub fn is_local(&self) -> bool {
        matches!(self, PeerIdentity::Local { .. })
    }

    /// Peer UID, if this identity carries one.
    pub fn uid(&self) -> Option<u32> {
        match self {
            PeerIdentity::Local { uid, .. } => Some(*uid),
            _ => None,
        }
    }

    /// Peer PID, if this identity carries a meaningful one
    /// (`Some(-1)` is filtered to `None`).
    pub fn pid(&self) -> Option<i32> {
        match self {
            PeerIdentity::Local { pid, .. } if *pid >= 0 => Some(*pid),
            _ => None,
        }
    }
}

impl fmt::Display for PeerIdentity {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            PeerIdentity::Local { uid, pid } => write!(f, "local(uid={uid}, pid={pid})"),
            PeerIdentity::Vsock { cid } => {
                write!(f, "vsock(cid={cid}; routing only, NOT an ACL basis)")
            }
            PeerIdentity::Certificate(c) => {
                write!(
                    f,
                    "cert(subject={:?}, sha256={})",
                    c.subject(),
                    c.fingerprint_hex()
                )
            }
            // Make the security-relevant "no identity" state loud.
            PeerIdentity::Anonymous => {
                write!(
                    f,
                    "anonymous(NO peer identity — access control NOT possible)"
                )
            }
        }
    }
}

// --- Length-prefix framing shared by stream backends -----------------
//
// Wire shape: `u32 little-endian length | <length> body bytes`. No
// magic / self-sync — the length is authoritative and bounded by
// MAX_FRAME_LEN before allocation. The `mem` backend does not use this
// (a channel message is already a frame).

/// Write one length-prefixed frame to a blocking stream.
pub(crate) fn write_frame<W: Write>(w: &mut W, buf: &[u8]) -> RpcResult<()> {
    if buf.len() > MAX_FRAME_LEN {
        return Err(RpcError::FrameTooLarge {
            declared: buf.len(),
            max: MAX_FRAME_LEN,
        });
    }
    // Length prefix + body coalesced into ONE buffer / ONE `write_all`
    // so the 4-byte length and the body can never be spliced by a
    // concurrent writer (the 3-`write_all` form let two threads
    // interleave a frame irrecoverably). The cross-thread *correctness*
    // guarantee is the per-session connection lock (session.rs
    // `enter_connection`); this additionally keeps the lock-free
    // small-frame paths (e.g. a `DEC_STRONG` from `RpcProxy::drop`)
    // from ever emitting a half-frame.
    let mut framed = Vec::with_capacity(4 + buf.len());
    framed.extend_from_slice(&(buf.len() as u32).to_le_bytes());
    framed.extend_from_slice(buf);
    w.write_all(&framed)?;
    w.flush()?;
    Ok(())
}

/// `write_all`, but telling a failure that put **nothing** on the wire
/// apart from one that stopped part-way.
///
/// `write_all` reports the same error either way, and the session's
/// send-failure rule needs the difference: a frame that stopped part-way
/// left the peer a header it will complete out of whatever arrives next,
/// so its connection must be retired, while a frame that never started
/// left the stream frame-synchronized and its connection healthy. A send
/// deadline (`SO_SNDTIMEO`) that expires before the first byte is
/// therefore [`RpcError::Timeout`] — the value a read deadline that
/// consumed nothing already yields — and every other failure, at any
/// position, stays the transport error.
///
/// `w` must report partial progress honestly, as a socket does. An
/// adapter that hands the whole buffer to another all-or-nothing send does
/// not, and classifies at its own level instead: `tls` never reports this,
/// because a record its socket write dropped is gone from the sequence
/// whether or not a byte of it went out.
pub(crate) fn write_all_reporting<W: Write>(w: &mut W, buf: &[u8]) -> RpcResult<()> {
    let mut sent = 0;
    while sent < buf.len() {
        match w.write(&buf[sent..]) {
            Ok(0) => return Err(std::io::Error::from(ErrorKind::WriteZero).into()),
            Ok(n) => sent += n,
            Err(e) if e.kind() == ErrorKind::Interrupted => continue,
            Err(e) if sent == 0 && is_timeout(&e) => return Err(RpcError::Timeout),
            Err(e) => return Err(RpcError::from(e)),
        }
    }
    Ok(())
}

/// Read exactly `buf.len()` bytes for a *frame header*. Zero bytes
/// before any progress is a clean [`RpcError::EndOfStream`]; a partial
/// header then EOF is [`RpcError::Truncated`]. A transport that can tell
/// an unclean end apart ([`RpcError::UncleanEndOfStream`], TLS with no
/// `close_notify`) reports it as itself before any progress and as
/// `Truncated` after — mid-frame, the stream position is what is lost.
fn read_header<R: Read>(r: &mut R, buf: &mut [u8]) -> RpcResult<()> {
    let mut filled = 0;
    while filled < buf.len() {
        match r.read(&mut buf[filled..]) {
            Ok(0) => {
                return Err(if filled == 0 {
                    RpcError::EndOfStream
                } else {
                    RpcError::Truncated
                });
            }
            Ok(n) => filled += n,
            Err(e) if e.kind() == ErrorKind::Interrupted => continue,
            // A read deadline that elapses with nothing consumed is a
            // clean Timeout (stream still frame-synchronized); mid-
            // header it is a desync → Truncated.
            Err(e) if is_timeout(&e) => {
                return Err(if filled == 0 {
                    RpcError::Timeout
                } else {
                    RpcError::DeadlineMidFrame
                });
            }
            Err(e) if e.kind() == ErrorKind::UnexpectedEof && filled > 0 => {
                return Err(RpcError::Truncated);
            }
            // Past the first byte the header is committed, so a disconnect
            // has lost the position whichever kind it arrived as: the ones
            // `From<io::Error>` folds into `EndOfStream` (a reset among
            // them) do not mean "nothing was pending" here.
            Err(e) => {
                return Err(match RpcError::from(e) {
                    RpcError::EndOfStream if filled > 0 => RpcError::Truncated,
                    other => other,
                })
            }
        }
    }
    Ok(())
}

/// Whether an I/O operation failed on a deadline this end armed:
/// `SO_RCVTIMEO` for the framing readers, `SO_SNDTIMEO` for
/// [`write_all_reporting`] and for `tls`'s teardown control flush. On the
/// supported platforms both surface as `WouldBlock`; `TimedOut` is here
/// because the `Read`
/// adapters carry [`RpcError::Timeout`] across the `RpcError` ⇄
/// `io::Error` boundary with exactly that kind, and the framing readers
/// sit on top of both. The cost is that a raw socket's own `ETIMEDOUT` —
/// a peer whose host stopped answering, not a deadline of ours — is
/// `TimedOut` too and cannot be told apart here; what is decided on that
/// distinction (see [`SessionEnd::new`](super::SessionEnd)) is decided
/// from whether a deadline was armed at all.
pub(crate) fn is_timeout(e: &std::io::Error) -> bool {
    matches!(e.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut)
}

/// The socket backends' `shutdown`: a second call raises `ENOTCONN` on
/// macOS (Linux returns `Ok`), and the trait promises idempotence, so that
/// one is absorbed. Anything else is the diagnostic the trait documents.
pub(crate) fn absorb_already_shut(r: std::io::Result<()>) -> RpcResult<()> {
    match r {
        Ok(()) => Ok(()),
        Err(e) if e.kind() == ErrorKind::NotConnected => Ok(()),
        Err(e) => Err(e.into()),
    }
}

/// Read exactly `buf.len()` body bytes. The header was already
/// committed, so any short read has lost the stream position:
/// [`RpcError::Truncated`] if the stream ended, [`RpcError::DeadlineMidFrame`]
/// if a read deadline cut it ([`is_timeout`] cannot say whose — see its doc).
fn read_body<R: Read>(r: &mut R, buf: &mut [u8]) -> RpcResult<()> {
    let mut filled = 0;
    while filled < buf.len() {
        match r.read(&mut buf[filled..]) {
            Ok(0) => return Err(RpcError::Truncated),
            Ok(n) => filled += n,
            Err(e) if e.kind() == ErrorKind::Interrupted => continue,
            Err(e) if is_timeout(&e) => return Err(RpcError::DeadlineMidFrame),
            Err(e) if e.kind() == ErrorKind::UnexpectedEof => return Err(RpcError::Truncated),
            // The header is already consumed, so every disconnect here is a
            // cut frame — including the kinds `From<io::Error>` folds into
            // `EndOfStream` (`ConnectionReset` from a peer killed mid-frame).
            Err(e) => {
                return Err(match RpcError::from(e) {
                    RpcError::EndOfStream => RpcError::Truncated,
                    other => other,
                })
            }
        }
    }
    Ok(())
}

/// Read one length-prefixed frame from a blocking stream.
pub(crate) fn read_frame<R: Read>(r: &mut R) -> RpcResult<Vec<u8>> {
    let mut len_buf = [0u8; 4];
    read_header(r, &mut len_buf)?;
    let len = u32::from_le_bytes(len_buf) as usize;
    if len > MAX_FRAME_LEN {
        // Reject *before* allocating `len` bytes.
        return Err(RpcError::FrameTooLarge {
            declared: len,
            max: MAX_FRAME_LEN,
        });
    }
    let mut body = vec![0u8; len];
    read_body(r, &mut body)?;
    Ok(body)
}

/// Decode-only entrypoint for the `rpc_frame_decode` fuzz target and
/// the deterministic adversarial-input regression tests. Feeds
/// arbitrary bytes through the same deframing path
/// `recv_frame` uses. `#[doc(hidden)]`: not part of the supported API
/// surface (and absent entirely without the `rpc` feature).
#[cfg(any(test, feature = "fuzzing"))]
#[doc(hidden)]
pub fn __fuzz_decode_frame(input: &[u8]) -> RpcResult<Vec<u8>> {
    read_frame(&mut std::io::Cursor::new(input))
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn frame_roundtrip_over_cursor() {
        for size in [0usize, 1, 4, 64, 4096, 1 << 20] {
            let payload: Vec<u8> = (0..size).map(|i| (i % 251) as u8).collect();
            let mut buf = Vec::new();
            write_frame(&mut buf, &payload).expect("write");
            let got = read_frame(&mut std::io::Cursor::new(&buf)).expect("read");
            assert_eq!(got, payload, "roundtrip mismatch at size {size}");
        }
    }

    #[test]
    fn two_frames_back_to_back_preserve_order() {
        let mut buf = Vec::new();
        write_frame(&mut buf, b"first").unwrap();
        write_frame(&mut buf, b"second").unwrap();
        let mut cur = std::io::Cursor::new(&buf);
        assert_eq!(read_frame(&mut cur).unwrap(), b"first");
        assert_eq!(read_frame(&mut cur).unwrap(), b"second");
    }

    /// Deterministic adversarial cases — must reject without
    /// allocating, panicking, or looping (mirrors the fuzz target).
    #[test]
    fn hostile_frame_headers_are_rejected_safely() {
        // Declared u32::MAX, no body: rejected pre-allocation.
        let huge = u32::MAX.to_le_bytes();
        assert!(matches!(
            __fuzz_decode_frame(&huge),
            Err(RpcError::FrameTooLarge { .. })
        ));

        // Declared MAX_FRAME_LEN + 1.
        let over = ((MAX_FRAME_LEN + 1) as u32).to_le_bytes();
        assert!(matches!(
            __fuzz_decode_frame(&over),
            Err(RpcError::FrameTooLarge { .. })
        ));

        // Empty input: clean peer-closed, no header at all.
        assert!(matches!(
            __fuzz_decode_frame(&[]),
            Err(RpcError::EndOfStream)
        ));

        // Partial header (2 of 4 bytes): truncated, not a panic.
        assert!(matches!(
            __fuzz_decode_frame(&[1, 0]),
            Err(RpcError::Truncated)
        ));

        // Header says 8 bytes, only 3 present: truncated body.
        let mut framed = 8u32.to_le_bytes().to_vec();
        framed.extend_from_slice(&[1, 2, 3]);
        assert!(matches!(
            __fuzz_decode_frame(&framed),
            Err(RpcError::Truncated)
        ));

        // A run of zero-length frames must not spin or panic.
        let zeros = vec![0u8; 4 * 1000];
        let mut cur = std::io::Cursor::new(&zeros[..]);
        for _ in 0..1000 {
            assert_eq!(read_frame(&mut cur).unwrap(), Vec::<u8>::new());
        }
    }

    #[test]
    fn write_frame_rejects_oversize_payload() {
        // We don't actually allocate MAX+1; just check the guard via a
        // fake writer that would error if written to.
        struct Trap;
        impl Write for Trap {
            fn write(&mut self, _: &[u8]) -> std::io::Result<usize> {
                panic!("oversize payload must be rejected before any write");
            }
            fn flush(&mut self) -> std::io::Result<()> {
                Ok(())
            }
        }
        // Build a slice header claiming oversize without allocating it:
        // use a zero-filled Vec of MAX+1 only conceptually — instead
        // assert the boundary with a borrowed empty slice and a forged
        // length check by calling the guard logic directly.
        let big = vec![0u8; MAX_FRAME_LEN + 1];
        assert!(matches!(
            write_frame(&mut Trap, &big),
            Err(RpcError::FrameTooLarge { .. })
        ));
    }

    /// A send deadline is only frame-synchronized while nothing has gone
    /// out, and that is the whole difference the session's send-failure
    /// rule acts on: `Timeout` keeps the connection, a transport error
    /// retires it.
    ///
    /// **Mutant gate**: dropping the `sent == 0` guard makes the second
    /// case report `Timeout` as well, and a peer left half a frame keeps
    /// its slot.
    #[test]
    fn a_send_deadline_is_a_timeout_only_before_the_first_byte() {
        struct Stall(usize);
        impl Write for Stall {
            fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
                if self.0 == 0 {
                    return Err(std::io::Error::from(ErrorKind::WouldBlock));
                }
                let n = self.0.min(buf.len());
                self.0 -= n;
                Ok(n)
            }
            fn flush(&mut self) -> std::io::Result<()> {
                Ok(())
            }
        }

        assert!(matches!(
            write_all_reporting(&mut Stall(0), b"frame"),
            Err(RpcError::Timeout)
        ));
        assert!(
            matches!(
                write_all_reporting(&mut Stall(2), b"frame"),
                Err(RpcError::Io(ref e)) if e.kind() == ErrorKind::WouldBlock
            ),
            "a deadline past the first byte left a partial frame on the wire"
        );
        assert!(write_all_reporting(&mut Stall(5), b"frame").is_ok());
    }

    #[test]
    fn peer_identity_display_and_accessors() {
        let local = PeerIdentity::Local { uid: 1000, pid: 42 };
        assert!(local.is_local());
        assert_eq!(local.uid(), Some(1000));
        assert_eq!(local.pid(), Some(42));
        assert_eq!(format!("{local}"), "local(uid=1000, pid=42)");

        let no_pid = PeerIdentity::Local { uid: 0, pid: -1 };
        assert_eq!(no_pid.pid(), None, "-1 pid is reported as unavailable");

        let anon = PeerIdentity::Anonymous;
        assert!(!anon.is_local());
        assert_eq!(anon.uid(), None);
        assert!(
            format!("{anon}").contains("NO peer identity"),
            "Anonymous Display must make the missing-identity state loud"
        );
    }
}