rustferry_remote/
stdio.rs

1//! Strict, bounded envelopes for one-request worker stdio control planes.
2
3use std::{
4    collections::BTreeSet,
5    fmt,
6    io::{Read, Write},
7};
8
9use semver::Version;
10use serde::{Deserialize, Deserializer, Serialize, Serializer};
11
12use crate::{
13    CURRENT_PROTOCOL_VERSION, HandshakeRequest, HandshakeResponse, IosArtifactType,
14    ProtocolVersion, ProviderCapabilities, ProviderCheck, ProviderCheckStatus,
15    ProviderDoctorReport, ProviderDoctorRequest, ProviderFeature, SigningMode, SourceMode,
16};
17
18/// Current schema for the worker stdio request and response envelopes.
19pub const WORKER_STDIO_SCHEMA_VERSION: u32 = 1;
20
21/// Maximum accepted bytes for one worker stdio request.
22pub const MAX_WORKER_STDIO_REQUEST_BYTES: usize = 64 * 1024;
23
24/// Maximum emitted or accepted bytes for one worker stdio response.
25pub const MAX_WORKER_STDIO_RESPONSE_BYTES: usize = 1024 * 1024;
26
27const MAX_REQUIRED_FEATURES: usize = 64;
28const MAX_VERSION_BYTES: usize = 128;
29const MAX_PROVIDER_CHECKS: usize = 256;
30const MAX_CHECK_MESSAGE_BYTES: usize = 512;
31const MAX_CHECK_HELP_BYTES: usize = 1024;
32
33/// One strict, versioned worker stdio request.
34#[derive(Clone, Debug, Eq, PartialEq)]
35pub struct WorkerStdioRequestEnvelope {
36    /// Stdio envelope schema version.
37    pub schema_version: u32,
38    /// Requested read-only control-plane operation.
39    pub request: WorkerStdioRequest,
40}
41
42impl WorkerStdioRequestEnvelope {
43    /// Wrap one request in the current stdio schema.
44    #[must_use]
45    pub const fn new(request: WorkerStdioRequest) -> Self {
46        Self {
47            schema_version: WORKER_STDIO_SCHEMA_VERSION,
48            request,
49        }
50    }
51
52    fn validate(&self) -> Result<(), WorkerStdioCodecError> {
53        validate_schema_version(self.schema_version)?;
54        match &self.request {
55            WorkerStdioRequest::Handshake(request) => {
56                CURRENT_PROTOCOL_VERSION
57                    .negotiate(request.protocol_version)
58                    .map_err(|_| WorkerStdioCodecError::IncompatibleProtocolVersion {
59                        supported: CURRENT_PROTOCOL_VERSION,
60                        received: request.protocol_version,
61                    })?;
62                if request.required_features.len() > MAX_REQUIRED_FEATURES {
63                    return Err(WorkerStdioCodecError::InvalidRequest);
64                }
65                if request.client_version.to_string().len() > MAX_VERSION_BYTES {
66                    return Err(WorkerStdioCodecError::InvalidRequest);
67                }
68            }
69            WorkerStdioRequest::ProviderDoctor(request) => {
70                CURRENT_PROTOCOL_VERSION
71                    .negotiate(request.protocol_version)
72                    .map_err(|_| WorkerStdioCodecError::IncompatibleProtocolVersion {
73                        supported: CURRENT_PROTOCOL_VERSION,
74                        received: request.protocol_version,
75                    })?;
76                if !valid_identifier(&request.operation_id) {
77                    return Err(WorkerStdioCodecError::InvalidRequest);
78                }
79            }
80        }
81        Ok(())
82    }
83}
84
85impl Serialize for WorkerStdioRequestEnvelope {
86    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
87    where
88        S: Serializer,
89    {
90        RequestEnvelopeRef::from(self).serialize(serializer)
91    }
92}
93
94impl<'de> Deserialize<'de> for WorkerStdioRequestEnvelope {
95    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
96    where
97        D: Deserializer<'de>,
98    {
99        let envelope = StrictRequestEnvelope::deserialize(deserializer)?;
100        Ok(Self::from(envelope))
101    }
102}
103
104/// Supported read-only worker stdio operations.
105#[derive(Clone, Debug, Eq, PartialEq)]
106pub enum WorkerStdioRequest {
107    /// Negotiate protocol compatibility and worker capabilities.
108    Handshake(HandshakeRequest),
109    /// Inspect worker readiness without changing host configuration.
110    ProviderDoctor(ProviderDoctorRequest),
111}
112
113/// One strict, versioned worker stdio response.
114#[derive(Clone, Debug, Eq, PartialEq)]
115pub struct WorkerStdioResponseEnvelope {
116    /// Stdio envelope schema version.
117    pub schema_version: u32,
118    /// Typed worker response.
119    pub response: WorkerStdioResponse,
120}
121
122impl WorkerStdioResponseEnvelope {
123    /// Wrap one response in the current stdio schema.
124    #[must_use]
125    pub const fn new(response: WorkerStdioResponse) -> Self {
126        Self {
127            schema_version: WORKER_STDIO_SCHEMA_VERSION,
128            response,
129        }
130    }
131
132    /// Build a stable, secret-free error response.
133    #[must_use]
134    pub fn error(code: impl Into<String>, message: impl Into<String>, retryable: bool) -> Self {
135        Self::new(WorkerStdioResponse::Error(WorkerStdioErrorResponse {
136            code: code.into(),
137            message: message.into(),
138            retryable,
139        }))
140    }
141
142    fn validate(&self) -> Result<(), WorkerStdioCodecError> {
143        validate_schema_version(self.schema_version)?;
144        match &self.response {
145            WorkerStdioResponse::Handshake(response) => {
146                CURRENT_PROTOCOL_VERSION
147                    .negotiate(response.protocol_version)
148                    .map_err(|_| WorkerStdioCodecError::IncompatibleProtocolVersion {
149                        supported: CURRENT_PROTOCOL_VERSION,
150                        received: response.protocol_version,
151                    })?;
152                if !valid_identifier(&response.provider) || !valid_identifier(&response.worker_id) {
153                    return Err(WorkerStdioCodecError::InvalidResponse);
154                }
155                if response.worker_version.to_string().len() > MAX_VERSION_BYTES {
156                    return Err(WorkerStdioCodecError::InvalidResponse);
157                }
158            }
159            WorkerStdioResponse::ProviderDoctor(response) => {
160                CURRENT_PROTOCOL_VERSION
161                    .negotiate(response.protocol_version)
162                    .map_err(|_| WorkerStdioCodecError::IncompatibleProtocolVersion {
163                        supported: CURRENT_PROTOCOL_VERSION,
164                        received: response.protocol_version,
165                    })?;
166                if !valid_identifier(&response.provider) {
167                    return Err(WorkerStdioCodecError::InvalidResponse);
168                }
169                if response.checks.len() > MAX_PROVIDER_CHECKS
170                    || response.checks.iter().any(|check| {
171                        !valid_identifier(&check.code)
172                            || !valid_public_text(&check.message, MAX_CHECK_MESSAGE_BYTES)
173                            || check
174                                .help
175                                .as_deref()
176                                .is_some_and(|help| !valid_public_text(help, MAX_CHECK_HELP_BYTES))
177                    })
178                    || response.ready
179                        != response
180                            .checks
181                            .iter()
182                            .all(|check| check.status != ProviderCheckStatus::Error)
183                {
184                    return Err(WorkerStdioCodecError::InvalidResponse);
185                }
186            }
187            WorkerStdioResponse::Error(error) => {
188                if !valid_identifier(&error.code)
189                    || !valid_public_text(&error.message, MAX_CHECK_MESSAGE_BYTES)
190                {
191                    return Err(WorkerStdioCodecError::InvalidResponse);
192                }
193            }
194        }
195        Ok(())
196    }
197}
198
199impl Serialize for WorkerStdioResponseEnvelope {
200    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
201    where
202        S: Serializer,
203    {
204        ResponseEnvelopeRef::from(self).serialize(serializer)
205    }
206}
207
208impl<'de> Deserialize<'de> for WorkerStdioResponseEnvelope {
209    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
210    where
211        D: Deserializer<'de>,
212    {
213        let envelope = StrictResponseEnvelope::deserialize(deserializer)?;
214        Ok(Self::from(envelope))
215    }
216}
217
218/// Supported worker stdio results.
219#[derive(Clone, Debug, Eq, PartialEq)]
220pub enum WorkerStdioResponse {
221    /// Negotiated worker identity and capabilities.
222    Handshake(HandshakeResponse),
223    /// Read-only provider readiness report.
224    ProviderDoctor(ProviderDoctorReport),
225    /// Stable public failure without raw parser or subprocess text.
226    Error(WorkerStdioErrorResponse),
227}
228
229/// Stable secret-free stdio failure payload.
230#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
231#[serde(deny_unknown_fields)]
232pub struct WorkerStdioErrorResponse {
233    /// Stable machine-readable error code.
234    pub code: String,
235    /// Bounded public summary.
236    pub message: String,
237    /// Whether retrying without configuration changes may succeed.
238    pub retryable: bool,
239}
240
241/// Sanitized failure returned by the bounded stdio codec.
242#[derive(Clone, Copy, Debug, Eq, PartialEq)]
243pub enum WorkerStdioCodecError {
244    /// Reading or writing the transport failed.
245    Io,
246    /// No JSON document was supplied.
247    EmptyInput,
248    /// The input ended inside a JSON document.
249    TruncatedJson,
250    /// JSON was malformed or did not match the strict envelope.
251    MalformedJson,
252    /// The request exceeded its fixed byte bound.
253    RequestTooLarge,
254    /// The response exceeded its fixed byte bound.
255    ResponseTooLarge,
256    /// The stdio envelope schema is unsupported.
257    UnsupportedSchemaVersion {
258        /// Supported schema version.
259        supported: u32,
260        /// Received schema version.
261        received: u32,
262    },
263    /// The remote-build protocol major version is incompatible.
264    IncompatibleProtocolVersion {
265        /// Supported protocol version.
266        supported: ProtocolVersion,
267        /// Received protocol version.
268        received: ProtocolVersion,
269    },
270    /// A decoded request violated a bounded semantic invariant.
271    InvalidRequest,
272    /// A decoded response violated a bounded semantic invariant.
273    InvalidResponse,
274}
275
276impl WorkerStdioCodecError {
277    /// Stable machine-readable error code.
278    #[must_use]
279    pub const fn code(self) -> &'static str {
280        match self {
281            Self::Io => "stdio_io_failed",
282            Self::EmptyInput => "empty_request",
283            Self::TruncatedJson => "truncated_json",
284            Self::MalformedJson => "malformed_json",
285            Self::RequestTooLarge => "request_too_large",
286            Self::ResponseTooLarge => "response_too_large",
287            Self::UnsupportedSchemaVersion { .. } => "unsupported_schema_version",
288            Self::IncompatibleProtocolVersion { .. } => "incompatible_protocol_version",
289            Self::InvalidRequest => "invalid_request",
290            Self::InvalidResponse => "invalid_response",
291        }
292    }
293
294    /// Stable public message which never contains input or parser text.
295    #[must_use]
296    pub const fn public_message(self) -> &'static str {
297        match self {
298            Self::Io => "worker stdio transport failed",
299            Self::EmptyInput => "worker request is empty",
300            Self::TruncatedJson => "worker request ended before one JSON document was complete",
301            Self::MalformedJson => "worker request is not a valid strict JSON envelope",
302            Self::RequestTooLarge => "worker request exceeds the accepted byte limit",
303            Self::ResponseTooLarge => "worker response exceeds the accepted byte limit",
304            Self::UnsupportedSchemaVersion { .. } => "worker stdio schema version is unsupported",
305            Self::IncompatibleProtocolVersion { .. } => {
306                "worker and client protocol versions are incompatible"
307            }
308            Self::InvalidRequest => "worker request failed semantic validation",
309            Self::InvalidResponse => "worker response failed semantic validation",
310        }
311    }
312}
313
314impl fmt::Display for WorkerStdioCodecError {
315    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
316        formatter.write_str(self.public_message())
317    }
318}
319
320impl std::error::Error for WorkerStdioCodecError {}
321
322/// Decode exactly one bounded request document from a reader.
323///
324/// # Errors
325///
326/// Returns a sanitized error for I/O failure, size overflow, malformed or
327/// truncated JSON, unknown fields, schema mismatch, or protocol mismatch.
328pub fn decode_worker_stdio_request(
329    reader: &mut impl Read,
330) -> Result<WorkerStdioRequestEnvelope, WorkerStdioCodecError> {
331    let bytes = read_bounded(reader, MAX_WORKER_STDIO_REQUEST_BYTES, true)?;
332    let envelope: WorkerStdioRequestEnvelope = decode_json(&bytes)?;
333    envelope.validate()?;
334    Ok(envelope)
335}
336
337/// Encode and write exactly one bounded request document plus a newline.
338///
339/// # Errors
340///
341/// Returns a sanitized error for invalid request data, serialization failure,
342/// size overflow, or transport failure.
343pub fn encode_worker_stdio_request(
344    writer: &mut impl Write,
345    request: &WorkerStdioRequestEnvelope,
346) -> Result<(), WorkerStdioCodecError> {
347    request.validate()?;
348    let mut encoded = BoundedOutput::new(MAX_WORKER_STDIO_REQUEST_BYTES);
349    if serde_json::to_writer(&mut encoded, request).is_err() {
350        return Err(if encoded.exceeded {
351            WorkerStdioCodecError::RequestTooLarge
352        } else {
353            WorkerStdioCodecError::InvalidRequest
354        });
355    }
356    encoded
357        .write_all(b"\n")
358        .map_err(|_| WorkerStdioCodecError::RequestTooLarge)?;
359    writer
360        .write_all(&encoded.bytes)
361        .map_err(|_| WorkerStdioCodecError::Io)
362}
363
364/// Encode and write exactly one bounded response document plus a newline.
365///
366/// # Errors
367///
368/// Returns a sanitized error for invalid response data, serialization failure,
369/// size overflow, or transport failure.
370pub fn encode_worker_stdio_response(
371    writer: &mut impl Write,
372    response: &WorkerStdioResponseEnvelope,
373) -> Result<(), WorkerStdioCodecError> {
374    response.validate()?;
375    let mut encoded = BoundedOutput::new(MAX_WORKER_STDIO_RESPONSE_BYTES);
376    if serde_json::to_writer(&mut encoded, response).is_err() {
377        return Err(if encoded.exceeded {
378            WorkerStdioCodecError::ResponseTooLarge
379        } else {
380            WorkerStdioCodecError::InvalidResponse
381        });
382    }
383    encoded
384        .write_all(b"\n")
385        .map_err(|_| WorkerStdioCodecError::ResponseTooLarge)?;
386    writer
387        .write_all(&encoded.bytes)
388        .map_err(|_| WorkerStdioCodecError::Io)
389}
390
391/// Decode exactly one bounded response document from a reader.
392///
393/// # Errors
394///
395/// Returns a sanitized error for I/O failure, size overflow, malformed or
396/// truncated JSON, unknown fields, schema mismatch, or protocol mismatch.
397pub fn decode_worker_stdio_response(
398    reader: &mut impl Read,
399) -> Result<WorkerStdioResponseEnvelope, WorkerStdioCodecError> {
400    let bytes = read_bounded(reader, MAX_WORKER_STDIO_RESPONSE_BYTES, false)?;
401    let envelope: WorkerStdioResponseEnvelope = decode_json(&bytes)?;
402    envelope.validate()?;
403    Ok(envelope)
404}
405
406fn read_bounded(
407    reader: &mut impl Read,
408    maximum: usize,
409    request: bool,
410) -> Result<Vec<u8>, WorkerStdioCodecError> {
411    let limit = u64::try_from(maximum.saturating_add(1)).unwrap_or(u64::MAX);
412    let mut bytes = Vec::with_capacity(maximum.min(8192));
413    reader
414        .take(limit)
415        .read_to_end(&mut bytes)
416        .map_err(|_| WorkerStdioCodecError::Io)?;
417    if bytes.len() > maximum {
418        return Err(if request {
419            WorkerStdioCodecError::RequestTooLarge
420        } else {
421            WorkerStdioCodecError::ResponseTooLarge
422        });
423    }
424    if bytes.iter().all(u8::is_ascii_whitespace) {
425        return Err(WorkerStdioCodecError::EmptyInput);
426    }
427    Ok(bytes)
428}
429
430fn decode_json<T>(bytes: &[u8]) -> Result<T, WorkerStdioCodecError>
431where
432    T: for<'de> Deserialize<'de>,
433{
434    serde_json::from_slice(bytes).map_err(|error| {
435        if error.is_eof() {
436            WorkerStdioCodecError::TruncatedJson
437        } else {
438            WorkerStdioCodecError::MalformedJson
439        }
440    })
441}
442
443const fn validate_schema_version(version: u32) -> Result<(), WorkerStdioCodecError> {
444    if version == WORKER_STDIO_SCHEMA_VERSION {
445        Ok(())
446    } else {
447        Err(WorkerStdioCodecError::UnsupportedSchemaVersion {
448            supported: WORKER_STDIO_SCHEMA_VERSION,
449            received: version,
450        })
451    }
452}
453
454fn valid_identifier(value: &str) -> bool {
455    !value.is_empty()
456        && value.len() <= 160
457        && value
458            .bytes()
459            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b':'))
460}
461
462fn valid_public_text(value: &str, maximum_bytes: usize) -> bool {
463    !value.is_empty() && value.len() <= maximum_bytes && !value.chars().any(char::is_control)
464}
465
466struct BoundedOutput {
467    bytes: Vec<u8>,
468    maximum: usize,
469    exceeded: bool,
470}
471
472impl BoundedOutput {
473    fn new(maximum: usize) -> Self {
474        Self {
475            bytes: Vec::with_capacity(maximum.min(8192)),
476            maximum,
477            exceeded: false,
478        }
479    }
480}
481
482impl Write for BoundedOutput {
483    fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
484        if bytes.len() > self.maximum.saturating_sub(self.bytes.len()) {
485            self.exceeded = true;
486            return Err(std::io::Error::other("bounded stdio output exceeded"));
487        }
488        self.bytes.extend_from_slice(bytes);
489        Ok(bytes.len())
490    }
491
492    fn flush(&mut self) -> std::io::Result<()> {
493        Ok(())
494    }
495}
496
497#[derive(Deserialize)]
498#[serde(deny_unknown_fields)]
499struct StrictRequestEnvelope {
500    schema_version: u32,
501    request: StrictRequest,
502}
503
504#[derive(Deserialize)]
505#[serde(
506    deny_unknown_fields,
507    rename_all = "snake_case",
508    tag = "operation",
509    content = "payload"
510)]
511enum StrictRequest {
512    Handshake(StrictHandshakeRequest),
513    ProviderDoctor(StrictProviderDoctorRequest),
514}
515
516#[derive(Deserialize)]
517#[serde(deny_unknown_fields)]
518struct StrictHandshakeRequest {
519    protocol_version: StrictProtocolVersion,
520    client_version: Version,
521    required_features: Vec<StrictProviderFeature>,
522}
523
524#[derive(Deserialize)]
525#[serde(deny_unknown_fields)]
526struct StrictProviderDoctorRequest {
527    protocol_version: StrictProtocolVersion,
528    operation_id: String,
529    require_signing: bool,
530}
531
532#[derive(Clone, Copy, Deserialize)]
533#[serde(deny_unknown_fields)]
534struct StrictProtocolVersion {
535    major: u16,
536    minor: u16,
537}
538
539impl From<StrictProtocolVersion> for ProtocolVersion {
540    fn from(version: StrictProtocolVersion) -> Self {
541        Self::new(version.major, version.minor)
542    }
543}
544
545#[derive(Deserialize)]
546#[serde(
547    deny_unknown_fields,
548    rename_all = "snake_case",
549    tag = "feature",
550    content = "value"
551)]
552enum StrictProviderFeature {
553    SourceMode(SourceMode),
554    IosDeviceBuild,
555    IosSimulatorBuild,
556    SigningMode(SigningMode),
557    PersonalTeam,
558    LiveEvents,
559    LiveLogs,
560    Cancellation,
561    ArtifactType(IosArtifactType),
562    Cache,
563    ArtifactListing,
564    ArtifactDownload,
565    Cleanup,
566    PhysicalDeviceAccess,
567}
568
569impl From<StrictProviderFeature> for ProviderFeature {
570    fn from(feature: StrictProviderFeature) -> Self {
571        match feature {
572            StrictProviderFeature::SourceMode(value) => Self::SourceMode(value),
573            StrictProviderFeature::IosDeviceBuild => Self::IosDeviceBuild,
574            StrictProviderFeature::IosSimulatorBuild => Self::IosSimulatorBuild,
575            StrictProviderFeature::SigningMode(value) => Self::SigningMode(value),
576            StrictProviderFeature::PersonalTeam => Self::PersonalTeam,
577            StrictProviderFeature::LiveEvents => Self::LiveEvents,
578            StrictProviderFeature::LiveLogs => Self::LiveLogs,
579            StrictProviderFeature::Cancellation => Self::Cancellation,
580            StrictProviderFeature::ArtifactType(value) => Self::ArtifactType(value),
581            StrictProviderFeature::Cache => Self::Cache,
582            StrictProviderFeature::ArtifactListing => Self::ArtifactListing,
583            StrictProviderFeature::ArtifactDownload => Self::ArtifactDownload,
584            StrictProviderFeature::Cleanup => Self::Cleanup,
585            StrictProviderFeature::PhysicalDeviceAccess => Self::PhysicalDeviceAccess,
586        }
587    }
588}
589
590impl From<StrictRequestEnvelope> for WorkerStdioRequestEnvelope {
591    fn from(envelope: StrictRequestEnvelope) -> Self {
592        let request = match envelope.request {
593            StrictRequest::Handshake(request) => WorkerStdioRequest::Handshake(HandshakeRequest {
594                protocol_version: request.protocol_version.into(),
595                client_version: request.client_version,
596                required_features: request
597                    .required_features
598                    .into_iter()
599                    .map(ProviderFeature::from)
600                    .collect(),
601            }),
602            StrictRequest::ProviderDoctor(request) => {
603                WorkerStdioRequest::ProviderDoctor(ProviderDoctorRequest {
604                    protocol_version: request.protocol_version.into(),
605                    operation_id: request.operation_id,
606                    require_signing: request.require_signing,
607                })
608            }
609        };
610        Self {
611            schema_version: envelope.schema_version,
612            request,
613        }
614    }
615}
616
617#[derive(Serialize)]
618struct RequestEnvelopeRef<'a> {
619    schema_version: u32,
620    request: RequestRef<'a>,
621}
622
623#[derive(Serialize)]
624#[serde(rename_all = "snake_case", tag = "operation", content = "payload")]
625enum RequestRef<'a> {
626    Handshake(&'a HandshakeRequest),
627    ProviderDoctor(&'a ProviderDoctorRequest),
628}
629
630impl<'a> From<&'a WorkerStdioRequestEnvelope> for RequestEnvelopeRef<'a> {
631    fn from(envelope: &'a WorkerStdioRequestEnvelope) -> Self {
632        let request = match &envelope.request {
633            WorkerStdioRequest::Handshake(request) => RequestRef::Handshake(request),
634            WorkerStdioRequest::ProviderDoctor(request) => RequestRef::ProviderDoctor(request),
635        };
636        Self {
637            schema_version: envelope.schema_version,
638            request,
639        }
640    }
641}
642
643#[derive(Deserialize)]
644#[serde(deny_unknown_fields)]
645struct StrictResponseEnvelope {
646    schema_version: u32,
647    response: StrictResponse,
648}
649
650#[derive(Deserialize)]
651#[serde(
652    deny_unknown_fields,
653    rename_all = "snake_case",
654    tag = "operation",
655    content = "payload"
656)]
657enum StrictResponse {
658    Handshake(StrictHandshakeResponse),
659    ProviderDoctor(StrictProviderDoctorReport),
660    Error(WorkerStdioErrorResponse),
661}
662
663#[derive(Deserialize)]
664#[serde(deny_unknown_fields)]
665struct StrictHandshakeResponse {
666    protocol_version: StrictProtocolVersion,
667    worker_version: Version,
668    provider: String,
669    worker_id: String,
670    capabilities: StrictProviderCapabilities,
671}
672
673#[derive(Deserialize)]
674#[serde(deny_unknown_fields)]
675struct StrictProviderDoctorReport {
676    protocol_version: StrictProtocolVersion,
677    provider: String,
678    ready: bool,
679    checks: Vec<StrictProviderCheck>,
680    capabilities: StrictProviderCapabilities,
681}
682
683#[derive(Deserialize)]
684#[serde(deny_unknown_fields)]
685struct StrictProviderCheck {
686    code: String,
687    status: ProviderCheckStatus,
688    message: String,
689    help: Option<String>,
690}
691
692#[derive(Deserialize)]
693#[serde(deny_unknown_fields)]
694#[allow(clippy::struct_excessive_bools)]
695struct StrictProviderCapabilities {
696    source_modes: BTreeSet<SourceMode>,
697    ios_device_build: bool,
698    ios_simulator_build: bool,
699    signing_modes: BTreeSet<SigningMode>,
700    personal_team: bool,
701    live_events: bool,
702    live_logs: bool,
703    cancellation: bool,
704    artifact_types: BTreeSet<IosArtifactType>,
705    cache: bool,
706    max_source_bytes: Option<u64>,
707    retention_seconds: Option<u64>,
708    artifact_listing: bool,
709    artifact_download: bool,
710    cleanup: bool,
711    physical_device_access: bool,
712}
713
714impl From<StrictProviderCapabilities> for ProviderCapabilities {
715    fn from(capabilities: StrictProviderCapabilities) -> Self {
716        Self {
717            source_modes: capabilities.source_modes,
718            ios_device_build: capabilities.ios_device_build,
719            ios_simulator_build: capabilities.ios_simulator_build,
720            signing_modes: capabilities.signing_modes,
721            personal_team: capabilities.personal_team,
722            live_events: capabilities.live_events,
723            live_logs: capabilities.live_logs,
724            cancellation: capabilities.cancellation,
725            artifact_types: capabilities.artifact_types,
726            cache: capabilities.cache,
727            max_source_bytes: capabilities.max_source_bytes,
728            retention_seconds: capabilities.retention_seconds,
729            artifact_listing: capabilities.artifact_listing,
730            artifact_download: capabilities.artifact_download,
731            cleanup: capabilities.cleanup,
732            physical_device_access: capabilities.physical_device_access,
733        }
734    }
735}
736
737impl From<StrictResponseEnvelope> for WorkerStdioResponseEnvelope {
738    fn from(envelope: StrictResponseEnvelope) -> Self {
739        let response = match envelope.response {
740            StrictResponse::Handshake(response) => {
741                WorkerStdioResponse::Handshake(HandshakeResponse {
742                    protocol_version: response.protocol_version.into(),
743                    worker_version: response.worker_version,
744                    provider: response.provider,
745                    worker_id: response.worker_id,
746                    capabilities: response.capabilities.into(),
747                })
748            }
749            StrictResponse::ProviderDoctor(response) => {
750                WorkerStdioResponse::ProviderDoctor(ProviderDoctorReport {
751                    protocol_version: response.protocol_version.into(),
752                    provider: response.provider,
753                    ready: response.ready,
754                    checks: response
755                        .checks
756                        .into_iter()
757                        .map(|check| ProviderCheck {
758                            code: check.code,
759                            status: check.status,
760                            message: check.message,
761                            help: check.help,
762                        })
763                        .collect(),
764                    capabilities: response.capabilities.into(),
765                })
766            }
767            StrictResponse::Error(error) => WorkerStdioResponse::Error(error),
768        };
769        Self {
770            schema_version: envelope.schema_version,
771            response,
772        }
773    }
774}
775
776#[derive(Serialize)]
777struct ResponseEnvelopeRef<'a> {
778    schema_version: u32,
779    response: ResponseRef<'a>,
780}
781
782#[derive(Serialize)]
783#[serde(rename_all = "snake_case", tag = "operation", content = "payload")]
784enum ResponseRef<'a> {
785    Handshake(&'a HandshakeResponse),
786    ProviderDoctor(&'a ProviderDoctorReport),
787    Error(&'a WorkerStdioErrorResponse),
788}
789
790impl<'a> From<&'a WorkerStdioResponseEnvelope> for ResponseEnvelopeRef<'a> {
791    fn from(envelope: &'a WorkerStdioResponseEnvelope) -> Self {
792        let response = match &envelope.response {
793            WorkerStdioResponse::Handshake(response) => ResponseRef::Handshake(response),
794            WorkerStdioResponse::ProviderDoctor(response) => ResponseRef::ProviderDoctor(response),
795            WorkerStdioResponse::Error(error) => ResponseRef::Error(error),
796        };
797        Self {
798            schema_version: envelope.schema_version,
799            response,
800        }
801    }
802}
803
804#[cfg(test)]
805mod tests {
806    use std::io::Cursor;
807
808    use super::*;
809
810    fn handshake_json() -> Vec<u8> {
811        br#"{"schema_version":1,"request":{"operation":"handshake","payload":{"protocol_version":{"major":1,"minor":0},"client_version":"0.1.0","required_features":[]}}}"#.to_vec()
812    }
813
814    #[test]
815    fn strict_request_roundtrip() {
816        let request = decode_worker_stdio_request(&mut Cursor::new(handshake_json()))
817            .expect("valid handshake request");
818        assert!(matches!(request.request, WorkerStdioRequest::Handshake(_)));
819
820        let response =
821            WorkerStdioResponseEnvelope::error("not_ready", "worker is not ready", false);
822        let mut encoded = Vec::new();
823        encode_worker_stdio_response(&mut encoded, &response).expect("encode response");
824        let decoded =
825            decode_worker_stdio_response(&mut Cursor::new(encoded)).expect("decode response");
826        assert_eq!(decoded, response);
827    }
828
829    #[test]
830    fn unknown_nested_fields_are_rejected() {
831        let input = br#"{"schema_version":1,"request":{"operation":"provider_doctor","payload":{"protocol_version":{"major":1,"minor":0,"future":true},"operation_id":"op-1","require_signing":false}}}"#;
832        assert_eq!(
833            decode_worker_stdio_request(&mut Cursor::new(input)),
834            Err(WorkerStdioCodecError::MalformedJson)
835        );
836    }
837
838    #[test]
839    fn truncated_and_oversized_requests_are_distinct() {
840        let truncated = br#"{"schema_version":1,"request":{"operation":"handshake""#;
841        assert_eq!(
842            decode_worker_stdio_request(&mut Cursor::new(truncated)),
843            Err(WorkerStdioCodecError::TruncatedJson)
844        );
845
846        let oversized = vec![b' '; MAX_WORKER_STDIO_REQUEST_BYTES + 1];
847        assert_eq!(
848            decode_worker_stdio_request(&mut Cursor::new(oversized)),
849            Err(WorkerStdioCodecError::RequestTooLarge)
850        );
851    }
852
853    #[test]
854    fn protocol_and_schema_mismatch_are_rejected() {
855        let protocol = br#"{"schema_version":1,"request":{"operation":"provider_doctor","payload":{"protocol_version":{"major":2,"minor":0},"operation_id":"op-1","require_signing":false}}}"#;
856        assert!(matches!(
857            decode_worker_stdio_request(&mut Cursor::new(protocol)),
858            Err(WorkerStdioCodecError::IncompatibleProtocolVersion { .. })
859        ));
860
861        let schema = br#"{"schema_version":2,"request":{"operation":"provider_doctor","payload":{"protocol_version":{"major":1,"minor":0},"operation_id":"op-1","require_signing":false}}}"#;
862        assert_eq!(
863            decode_worker_stdio_request(&mut Cursor::new(schema)),
864            Err(WorkerStdioCodecError::UnsupportedSchemaVersion {
865                supported: 1,
866                received: 2,
867            })
868        );
869    }
870
871    #[test]
872    fn hostile_doctor_text_and_unknown_response_fields_are_rejected() {
873        let hostile = br#"{"schema_version":1,"response":{"operation":"provider_doctor","payload":{"protocol_version":{"major":1,"minor":0},"provider":"ssh-macos","ready":true,"checks":[{"code":"host.macos","status":"ready","message":"bad\u001b[2J"}],"capabilities":{"source_modes":[],"ios_device_build":false,"ios_simulator_build":false,"signing_modes":[],"personal_team":false,"live_events":false,"live_logs":false,"cancellation":false,"artifact_types":[],"cache":false,"max_source_bytes":null,"retention_seconds":null,"artifact_listing":false,"artifact_download":false,"cleanup":false,"physical_device_access":false}}}}"#;
874        assert_eq!(
875            decode_worker_stdio_response(&mut Cursor::new(hostile)),
876            Err(WorkerStdioCodecError::InvalidResponse)
877        );
878
879        let unknown = br#"{"schema_version":1,"response":{"operation":"error","payload":{"code":"failed","message":"request failed","retryable":false,"raw_stderr":"private"}}}"#;
880        assert_eq!(
881            decode_worker_stdio_response(&mut Cursor::new(unknown)),
882            Err(WorkerStdioCodecError::MalformedJson)
883        );
884    }
885
886    #[test]
887    fn invalid_oversized_response_is_rejected_before_write() {
888        let response = WorkerStdioResponseEnvelope::error(
889            "oversized",
890            "x".repeat(MAX_WORKER_STDIO_RESPONSE_BYTES),
891            false,
892        );
893        let mut output = Vec::new();
894        assert_eq!(
895            encode_worker_stdio_response(&mut output, &response),
896            Err(WorkerStdioCodecError::InvalidResponse)
897        );
898        assert!(output.is_empty());
899    }
900
901    #[test]
902    fn bounded_output_never_retains_bytes_beyond_its_limit() {
903        let mut output = BoundedOutput::new(3);
904        assert!(output.write_all(b"four").is_err());
905        assert!(output.exceeded);
906        assert!(output.bytes.is_empty());
907    }
908}
rustferry_remote/stdio.rs

rustferry_remote/
stdio.rs
