lean_ctx/ipc/
runtime.rs

1// SPDX-License-Identifier: Apache-2.0
2//! Bounded client for an explicitly provisioned local Intelligence session.
3//! This module neither admits policy nor discovers, installs or trusts a peer.
4//! Callers own fresh scoped keys, sequence allocation, revocation and fallback.
5
6use std::fmt;
7use std::time::{Duration, SystemTime, UNIX_EPOCH};
8
9use lean_ctx_protocol::runtime_exchange::{
10    RuntimeExchangeProfile, RuntimeRequestV1, RuntimeResponseV1,
11};
12use lean_ctx_protocol::runtime_frame::{
13    RUNTIME_FRAME_HEADER_BYTES, RUNTIME_FRAME_NONCE_BYTES, RUNTIME_FRAME_TAG_BYTES,
14    RuntimeFrameDirection, decode_runtime_frame_in, encode_runtime_frame_in,
15    runtime_payload_length_in,
16};
17use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
18use tokio::time::{Instant, timeout_at};
19
20use super::DaemonAddr;
21
22/// Content-free failures: never expose a key, endpoint or projection in logs.
23#[derive(Debug, Clone, Copy, PartialEq, Eq)]
24pub enum RuntimeExchangeError {
25    InvalidRequest,
26    InvalidEndpoint,
27    EntropyUnavailable,
28    DeadlineExceeded,
29    ConnectionUnavailable,
30    Io,
31    InvalidFrame,
32    InvalidResponse,
33    ClockUnavailable,
34}
35
36impl fmt::Display for RuntimeExchangeError {
37    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
38        formatter.write_str(match self {
39            Self::InvalidRequest => "invalid local runtime request",
40            Self::InvalidEndpoint => "invalid local runtime endpoint",
41            Self::EntropyUnavailable => "local runtime randomness unavailable",
42            Self::DeadlineExceeded => "local runtime deadline exceeded",
43            Self::ConnectionUnavailable => "local runtime connection unavailable",
44            Self::Io => "local runtime I/O failed",
45            Self::InvalidFrame => "invalid authenticated runtime frame",
46            Self::InvalidResponse => "invalid local runtime response",
47            Self::ClockUnavailable => "local runtime clock unavailable",
48        })
49    }
50}
51
52impl std::error::Error for RuntimeExchangeError {}
53
54/// Exchange one admitted request over a fresh connection to an explicit endpoint.
55///
56/// The caller must provision and verify the endpoint and peer independently, use
57/// a fresh capability-scoped session key, serialize sequence allocation, and retire
58/// the session on failure/reconnect. A valid MAC is not policy authorization.
59/// No default daemon socket or global HTTP token is used. All connect/write/read
60/// work shares ONE monotonic budget; even Windows connection retries are bounded.
61/// The stream is dropped on every return. The caller alone persists host receipts
62/// and decides whether to use the deterministic public fallback.
63pub async fn exchange(
64    address: &DaemonAddr,
65    key: &[u8; 32],
66    request: &RuntimeRequestV1,
67) -> Result<RuntimeResponseV1, RuntimeExchangeError> {
68    exchange_in(address, key, request, RuntimeExchangeProfile::ORDINARY).await
69}
70
71/// [`exchange`] under a size profile the caller admitted from the
72/// authenticated package and the exact operation — never from the request,
73/// which could name any operation.
74pub(crate) async fn exchange_in(
75    address: &DaemonAddr,
76    key: &[u8; 32],
77    request: &RuntimeRequestV1,
78    profile: RuntimeExchangeProfile,
79) -> Result<RuntimeResponseV1, RuntimeExchangeError> {
80    validate_local_address(address)?;
81    let (deadline, frame) = prepare_exchange(key, request, profile)?;
82    let response = timeout_at(deadline, async {
83        let mut stream = super::connect(address)
84            .await
85            .map_err(|_| RuntimeExchangeError::ConnectionUnavailable)?;
86        exchange_connected(&mut stream, key, request, &frame, profile).await
87    })
88    .await
89    .map_err(|_| RuntimeExchangeError::DeadlineExceeded)?;
90    finish_exchange(deadline, response)
91}
92
93/// Exchange over a caller-protected connected stream, including inherited Unix
94/// streams and one-use Windows named pipes. Authentication and deadline checks
95/// are identical to the explicit-endpoint exchange path. The profile is the
96/// caller's own admission, as for [`exchange_in`].
97pub(crate) async fn exchange_inherited<S: AsyncRead + AsyncWrite + Unpin>(
98    stream: &mut S,
99    key: &[u8; 32],
100    request: &RuntimeRequestV1,
101    profile: RuntimeExchangeProfile,
102) -> Result<RuntimeResponseV1, RuntimeExchangeError> {
103    let (deadline, frame) = prepare_exchange(key, request, profile)?;
104    let response = timeout_at(
105        deadline,
106        exchange_connected(stream, key, request, &frame, profile),
107    )
108    .await
109    .map_err(|_| RuntimeExchangeError::DeadlineExceeded)?;
110    finish_exchange(deadline, response)
111}
112
113fn prepare_exchange(
114    key: &[u8; 32],
115    request: &RuntimeRequestV1,
116    profile: RuntimeExchangeProfile,
117) -> Result<(Instant, Vec<u8>), RuntimeExchangeError> {
118    let started = Instant::now();
119    let now = unix_millis()?;
120    request
121        .validate_at_in(now, profile)
122        .map_err(|_| RuntimeExchangeError::InvalidRequest)?;
123    let deadline = started + Duration::from_millis(request.deadline_unix_ms - now);
124    let payload = request
125        .to_bytes_in(profile)
126        .map_err(|_| RuntimeExchangeError::InvalidRequest)?;
127    let mut nonce = [0; RUNTIME_FRAME_NONCE_BYTES];
128    getrandom::fill(&mut nonce).map_err(|_| RuntimeExchangeError::EntropyUnavailable)?;
129    let frame = encode_runtime_frame_in(
130        key,
131        RuntimeFrameDirection::Request,
132        &nonce,
133        &payload,
134        profile,
135    )
136    .map_err(|_| RuntimeExchangeError::InvalidFrame)?;
137    ensure_remaining(deadline)?;
138    Ok((deadline, frame))
139}
140
141fn finish_exchange(
142    deadline: Instant,
143    response: Result<RuntimeResponseV1, RuntimeExchangeError>,
144) -> Result<RuntimeResponseV1, RuntimeExchangeError> {
145    // timeout_at can poll a ready future after its deadline; reject that too.
146    ensure_remaining(deadline)?;
147    response
148}
149
150fn validate_local_address(address: &DaemonAddr) -> Result<(), RuntimeExchangeError> {
151    let valid = match address {
152        #[cfg(unix)]
153        DaemonAddr::Unix(path) => path.is_absolute(),
154        #[cfg(windows)]
155        DaemonAddr::NamedPipe(name) => name.strip_prefix(r"\\.\pipe\").is_some_and(|suffix| {
156            !suffix.is_empty()
157                && suffix.len() <= 200
158                && !suffix
159                    .chars()
160                    .any(|ch| ch.is_control() || ch == '\\' || ch == '/')
161        }),
162    };
163    if !valid {
164        return Err(RuntimeExchangeError::InvalidEndpoint);
165    }
166    Ok(())
167}
168
169async fn exchange_connected<S: AsyncRead + AsyncWrite + Unpin>(
170    stream: &mut S,
171    key: &[u8; 32],
172    request: &RuntimeRequestV1,
173    frame: &[u8],
174    profile: RuntimeExchangeProfile,
175) -> Result<RuntimeResponseV1, RuntimeExchangeError> {
176    stream
177        .write_all(frame)
178        .await
179        .map_err(|_| RuntimeExchangeError::Io)?;
180    stream.flush().await.map_err(|_| RuntimeExchangeError::Io)?;
181    let response_frame = read_frame(stream, profile).await?;
182    let payload = decode_runtime_frame_in(
183        key,
184        RuntimeFrameDirection::Response,
185        &response_frame,
186        profile,
187    )
188    .map_err(|_| RuntimeExchangeError::InvalidFrame)?;
189    RuntimeResponseV1::from_bytes_in(&payload, request, unix_millis()?, profile)
190        .map_err(|_| RuntimeExchangeError::InvalidResponse)
191}
192
193async fn read_frame<S: AsyncRead + Unpin>(
194    stream: &mut S,
195    profile: RuntimeExchangeProfile,
196) -> Result<Vec<u8>, RuntimeExchangeError> {
197    let mut header = [0_u8; RUNTIME_FRAME_HEADER_BYTES];
198    stream
199        .read_exact(&mut header)
200        .await
201        .map_err(|_| RuntimeExchangeError::Io)?;
202    // Validate magic/version and the length limit before allocation or body reads.
203    let length = runtime_payload_length_in(&header, RuntimeFrameDirection::Response, profile)
204        .map_err(|_| RuntimeExchangeError::InvalidFrame)?;
205    let mut frame = vec![
206        0;
207        RUNTIME_FRAME_HEADER_BYTES
208            + RUNTIME_FRAME_NONCE_BYTES
209            + RUNTIME_FRAME_TAG_BYTES
210            + length
211    ];
212    frame[..RUNTIME_FRAME_HEADER_BYTES].copy_from_slice(&header);
213    stream
214        .read_exact(&mut frame[RUNTIME_FRAME_HEADER_BYTES..])
215        .await
216        .map_err(|_| RuntimeExchangeError::Io)?;
217    Ok(frame)
218}
219
220fn ensure_remaining(deadline: Instant) -> Result<(), RuntimeExchangeError> {
221    if Instant::now() >= deadline {
222        return Err(RuntimeExchangeError::DeadlineExceeded);
223    }
224    Ok(())
225}
226
227fn unix_millis() -> Result<u64, RuntimeExchangeError> {
228    SystemTime::now()
229        .duration_since(UNIX_EPOCH)
230        .ok()
231        .and_then(|duration| u64::try_from(duration.as_millis()).ok())
232        .ok_or(RuntimeExchangeError::ClockUnavailable)
233}
234
235#[cfg(test)]
236mod tests;
lean_ctx/ipc/runtime.rs

lean_ctx/ipc/
runtime.rs
