kunobi_daemon/
wire.rs

1//! Negotiated binary sessions on a separate v2 endpoint.
2//!
3//! Keep the legacy listener and its discovery key unchanged. Only clients that
4//! discover an explicitly advertised v2 endpoint may send this protocol. There
5//! is no fallback after starting negotiation or sending an application request.
6//! The caller authenticates the OS peer and applies I/O deadlines before using
7//! this module. A codec cannot enforce deadlines on an arbitrary blocking I/O.
8
9use buffa::Message;
10
11#[rustfmt::skip]
12#[allow(missing_docs, clippy::derivable_impls)]
13mod generated;
14use crate::ServiceIdentity;
15pub use generated::{
16    Control, Health, Hello, LifecycleChange, LifecycleEvent, MessageKind, Refusal,
17};
18use std::{
19    fmt,
20    io::{self, Read, Write},
21};
22
23/// Identifies the binary transport, independently of application versions.
24pub const MAGIC: [u8; 8] = *b"KNDPB002";
25/// First binary protocol version. Legacy application protocols are unchanged.
26pub const VERSION: u32 = 2;
27/// Absolute message bound, including encoded field overhead.
28pub const MAX_FRAME: u32 = 65_536;
29/// Bound used before a peer's receive limit is negotiated.
30pub const HELLO_LIMIT: u32 = 1_024;
31/// Smallest usable negotiated frame limit.
32pub const MIN_FRAME: u32 = 256;
33
34/// Capabilities agreed for one connection, not inferred from the build string.
35pub mod capability {
36    /// Read-only health observations.
37    pub const HEALTH: u64 = 1;
38    /// Close admission and drain active work.
39    pub const DRAIN: u64 = 2;
40    /// Cooperative prepare/ready/commit/abort exchange.
41    pub const HANDOFF: u64 = 4;
42    /// Application-defined operations, carried as opaque bounded payloads.
43    pub const APPLICATION: u64 = 8;
44    /// Typed in-memory Health payloads on HEALTH and DRAIN replies.
45    pub const HEALTH_DETAILS: u64 = 16;
46    /// A bounded stream of lifecycle snapshots and changes on WATCH.
47    pub const WATCH: u64 = 32;
48}
49
50impl Health {
51    /// Encode a typed observation, preserving the request correlation ID.
52    pub fn response(&self, request: &Control) -> io::Result<Control> {
53        if request.kind != MessageKind::Lifecycle
54            || !matches!(request.operation, operation::HEALTH | operation::DRAIN)
55        {
56            return Err(invalid("not a health or drain request"));
57        }
58        Ok(Control {
59            operation: request.operation,
60            request_id: request.request_id,
61            generation: self.generation,
62            payload: self.encode_to_vec(),
63            ..Default::default()
64        })
65    }
66
67    /// Decode a correlated reply. The caller must compare process/build policy.
68    pub fn from_response(reply: &Control, request: &Control) -> io::Result<Self> {
69        if reply.kind != MessageKind::Lifecycle
70            || !matches!(request.operation, operation::HEALTH | operation::DRAIN)
71            || reply.operation != request.operation
72            || reply.request_id != request.request_id
73            || !reply.token.is_empty()
74            || reply.offset.is_some()
75        {
76            return Err(invalid("unexpected health response"));
77        }
78        let health: Self = decode_message(&reply.payload, MAX_FRAME)?;
79        if health.process_id == 0
80            || health.generation != reply.generation
81            || (health.ready && health.draining)
82        {
83            return Err(invalid("invalid health observation"));
84        }
85        Ok(health)
86    }
87}
88
89/// Stable operation numbers. Never reuse retired numbers.
90pub mod operation {
91    /// Health request or response; application defines the observation payload.
92    pub const HEALTH: u32 = 1;
93    /// Drain request or acknowledgement.
94    pub const DRAIN: u32 = 2;
95    /// Offer a destination generation.
96    pub const PREPARE: u32 = 3;
97    /// Acknowledge the pause boundary.
98    pub const PAUSED: u32 = 4;
99    /// Acknowledge readiness to switch.
100    pub const READY: u32 = 5;
101    /// Commit the negotiated handoff.
102    pub const COMMIT: u32 = 6;
103    /// Abandon a handoff attempt.
104    pub const ABORT: u32 = 7;
105    /// Subscribe to lifecycle state; no further requests use this session.
106    pub const WATCH: u32 = 8;
107}
108
109impl Hello {
110    /// Offer capabilities for a stable service installation. Use the same
111    /// identity to derive resource paths and negotiate on both endpoints.
112    /// The offer reads refusals: [`Session::connect`] reports a server's
113    /// [`Refused`] reason.
114    pub fn new(identity: &ServiceIdentity, supported: u64, required: u64) -> Self {
115        Self {
116            minimum: VERSION,
117            maximum: VERSION,
118            supported,
119            required,
120            max_frame: MAX_FRAME,
121            application: identity.application().into(),
122            profile: identity.profile().into(),
123            instance: identity.instance().into(),
124            service_id: identity.id().to_vec(),
125            reads_refusal: true,
126            ..Default::default()
127        }
128    }
129
130    fn validate(&self) -> io::Result<()> {
131        ServiceIdentity::new(
132            self.service_id
133                .as_slice()
134                .try_into()
135                .map_err(|_| invalid("service UUID must have 16 bytes"))?,
136            &self.application,
137            &self.profile,
138            &self.instance,
139        )
140        .map_err(|_| invalid("invalid service identity"))?;
141        if !Terms::of(self).valid() {
142            return Err(invalid("invalid version or capability offer"));
143        }
144        Ok(())
145    }
146}
147
148/// The numeric part of an offer: what negotiation compares once both peers
149/// have shown the same service identity. Kept apart from the identity strings
150/// so the agreement rules can be model-checked for every value.
151#[derive(Clone, Copy, Debug, PartialEq, Eq)]
152#[cfg_attr(kani, derive(kani::Arbitrary))]
153struct Terms {
154    minimum: u32,
155    maximum: u32,
156    supported: u64,
157    required: u64,
158    max_frame: u32,
159}
160
161impl Terms {
162    fn of(hello: &Hello) -> Self {
163        Self {
164            minimum: hello.minimum,
165            maximum: hello.maximum,
166            supported: hello.supported,
167            required: hello.required,
168            max_frame: hello.max_frame,
169        }
170    }
171
172    fn valid(self) -> bool {
173        self.minimum != 0
174            && self.minimum <= self.maximum
175            && self.required & !self.supported == 0
176            && (MIN_FRAME..=MAX_FRAME).contains(&self.max_frame)
177    }
178
179    fn agree(self, remote: Self) -> io::Result<Agreement> {
180        let minimum = self.minimum.max(remote.minimum);
181        let maximum = self.maximum.min(remote.maximum);
182        if minimum > VERSION || maximum < VERSION {
183            return Err(invalid("no implemented binary version in common"));
184        }
185        let capabilities = self.supported & remote.supported;
186        if (self.required | remote.required) & !capabilities != 0 {
187            return Err(invalid("required capability unavailable"));
188        }
189        Ok(Agreement {
190            version: VERSION,
191            capabilities,
192            max_frame: self.max_frame.min(remote.max_frame),
193        })
194    }
195}
196
197/// Agreed limits for one connection. Both peers independently calculate this.
198#[derive(Clone, Copy, Debug, PartialEq, Eq)]
199pub struct Agreement {
200    /// Selected binary version.
201    pub version: u32,
202    /// Intersection of the peers' supported capabilities.
203    pub capabilities: u64,
204    /// Maximum encoded message size in either direction.
205    pub max_frame: u32,
206}
207
208/// Negotiate compatible versions, required capabilities and receive bounds.
209/// Unknown optional capabilities are ignored; unknown required ones fail.
210pub fn negotiate(local: &Hello, remote: &Hello) -> io::Result<Agreement> {
211    local.validate()?;
212    remote.validate()?;
213    if !same_service(local, remote) {
214        return Err(invalid("service identity mismatch"));
215    }
216    Terms::of(local).agree(Terms::of(remote))
217}
218
219fn same_service(a: &Hello, b: &Hello) -> bool {
220    (&a.service_id, &a.application, &a.profile, &a.instance)
221        == (&b.service_id, &b.application, &b.profile, &b.instance)
222}
223
224fn required_capability(message: &Control) -> io::Result<u64> {
225    match message.kind.as_known() {
226        Some(MessageKind::Application) if message.operation != 0 => {
227            return Ok(capability::APPLICATION);
228        }
229        Some(MessageKind::Lifecycle) => {}
230        _ => return Err(invalid("invalid message kind or application operation")),
231    }
232    use self::{capability as c, operation as o};
233    match message.operation {
234        o::HEALTH => Ok(c::HEALTH),
235        o::DRAIN => Ok(c::DRAIN),
236        o::PREPARE..=o::ABORT => Ok(c::HANDOFF),
237        o::WATCH => Ok(c::WATCH),
238        _ => Err(invalid("unknown control operation")),
239    }
240}
241
242/// Why a server refused a connection during negotiation. A client handshake
243/// returns it inside an [`io::ErrorKind::ConnectionRefused`] error; read it
244/// with [`Refused::of`]. A server that closes without a reason, an older
245/// server included, still produces an ordinary I/O error.
246#[derive(Clone, Copy, Debug, PartialEq, Eq)]
247#[non_exhaustive]
248pub enum Refused {
249    /// The client lacks a binary version or capability the server requires.
250    /// The same client build cannot succeed against this server.
251    ProtocolTooOld,
252    /// The server's peer policy rejected this client.
253    PeerRejected,
254    /// The server is retiring. Read discovery again to find its successor.
255    Retiring,
256}
257
258impl Refused {
259    /// The refusal inside an error from a client handshake, if the server sent one.
260    pub fn of(error: &io::Error) -> Option<Self> {
261        error.get_ref()?.downcast_ref::<Self>().copied()
262    }
263
264    /// The frame refusing `remote`, if that client reads refusals and sends a
265    /// well-formed offer for the service `offer` serves. A wrong identity or a
266    /// malformed offer gets no reply, as from [`Session::accept`].
267    fn frame(self, offer: &Hello, remote: &Hello) -> Option<Hello> {
268        let reason = match self {
269            Self::ProtocolTooOld => Refusal::PROTOCOL_TOO_OLD,
270            Self::PeerRejected => Refusal::PEER_REJECTED,
271            Self::Retiring => Refusal::SERVER_RETIRING,
272        };
273        let answered = remote.reads_refusal && same_service(offer, remote);
274        (answered && remote.validate().is_ok()).then(|| Hello {
275            refusal: reason.into(),
276            ..Default::default()
277        })
278    }
279}
280
281/// What a server sends before closing on a failed negotiation: a refusal only
282/// to a client that lacks a version or capability `offer` requires. A server
283/// too old for the client gets only the close.
284fn rejection(offer: &Hello, remote: &Hello) -> Option<Hello> {
285    if remote.maximum < VERSION || offer.required & !remote.supported != 0 {
286        Refused::ProtocolTooOld.frame(offer, remote)
287    } else {
288        None
289    }
290}
291
292/// Fail with the server's refusal if `remote` is one rather than an offer.
293fn check_refusal(remote: &Hello) -> io::Result<()> {
294    let refused = match remote.refusal.as_known() {
295        Some(Refusal::NOT_REFUSED) => return Ok(()),
296        Some(Refusal::PROTOCOL_TOO_OLD) => Refused::ProtocolTooOld,
297        Some(Refusal::PEER_REJECTED) => Refused::PeerRejected,
298        Some(Refusal::SERVER_RETIRING) => Refused::Retiring,
299        None => {
300            return Err(io::Error::new(
301                io::ErrorKind::ConnectionRefused,
302                "connection refused for an unknown reason",
303            ));
304        }
305    };
306    Err(refused.into())
307}
308
309impl fmt::Display for Refused {
310    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
311        f.write_str(match self {
312            Self::ProtocolTooOld => "the server requires a newer protocol than this client offers",
313            Self::PeerRejected => "the server's peer policy rejected this client",
314            Self::Retiring => "the server is retiring; read discovery again",
315        })
316    }
317}
318
319impl std::error::Error for Refused {}
320
321impl From<Refused> for io::Error {
322    fn from(refused: Refused) -> Self {
323        io::Error::new(io::ErrorKind::ConnectionRefused, refused)
324    }
325}
326
327fn invalid(message: &'static str) -> io::Error {
328    io::Error::new(io::ErrorKind::InvalidData, message)
329}
330
331fn encode_frame<M: Message>(
332    message: &M,
333    limit: u32,
334    buffer: &mut Vec<u8>,
335    cache: &mut buffa::SizeCache,
336) -> io::Result<()> {
337    buffer.clear();
338    buffer.extend_from_slice(&[0; 4]);
339    let length = message
340        .try_encode_bounded_with_cache(limit, cache, buffer)
341        .map_err(io::Error::other)?;
342    if length == 0 {
343        return Err(invalid("empty message"));
344    }
345    buffer[..4].copy_from_slice(&length.to_le_bytes());
346    Ok(())
347}
348
349pub(crate) fn decode_message<M: Message>(buffer: &[u8], limit: u32) -> io::Result<M> {
350    buffa::DecodeOptions::new()
351        .with_max_message_size(limit as usize)
352        .with_recursion_limit(32)
353        .with_unknown_field_limit(128)
354        .with_element_memory_limit(MAX_FRAME as usize)
355        .decode_from_slice(buffer)
356        .map_err(io::Error::other)
357}
358
359/// Bounded framing with reusable scratch storage. The first four bytes are the
360/// encoded body length in little endian, not a native Rust struct layout.
361/// A framing/I/O error poisons this codec: reconnect instead of retrying bytes.
362pub struct Framed<S> {
363    io: S,
364    buffer: Vec<u8>,
365    cache: buffa::SizeCache,
366    limit: u32,
367    failed: bool,
368}
369
370impl<S: Read + Write> Framed<S> {
371    /// Construct a frame codec; normally use [`Session::connect`] or [`Session::accept`].
372    pub fn new(io: S, limit: u32) -> io::Result<Self> {
373        if !(MIN_FRAME..=MAX_FRAME).contains(&limit) {
374            return Err(invalid("invalid frame limit"));
375        }
376        Ok(Self {
377            io,
378            buffer: Vec::new(),
379            cache: buffa::SizeCache::new(),
380            limit,
381            failed: false,
382        })
383    }
384
385    fn usable(&self) -> io::Result<()> {
386        if self.failed {
387            Err(invalid("session failed; reconnect without replay"))
388        } else {
389            Ok(())
390        }
391    }
392
393    /// Encode one complete bounded message and flush it. No automatic retries.
394    pub fn send<M: Message>(&mut self, message: &M) -> io::Result<()> {
395        self.usable()?;
396        encode_frame(message, self.limit, &mut self.buffer, &mut self.cache)?;
397        self.failed = true;
398        self.io.write_all(&self.buffer)?;
399        self.io.flush()?;
400        self.failed = false;
401        Ok(())
402    }
403
404    /// Read exactly one message without consuming bytes from the next frame.
405    /// An oversized header is rejected before allocating its advertised size.
406    pub fn receive<M: Message + Default>(&mut self) -> io::Result<M> {
407        self.usable()?;
408        self.failed = true;
409        let mut length = [0; 4];
410        self.io.read_exact(&mut length)?;
411        let length = u32::from_le_bytes(length);
412        if length == 0 || length > self.limit {
413            return Err(invalid("received frame exceeds negotiated bounds"));
414        }
415        self.buffer.resize(length as usize, 0);
416        self.io.read_exact(&mut self.buffer)?;
417        let value = decode_message(self.buffer.as_slice(), self.limit)?;
418        self.failed = false;
419        Ok(value)
420    }
421}
422
423/// An established binary session. Construction completes both offers and both
424/// acceptance bytes before application operations may be sent or received.
425pub struct Session<S> {
426    frames: Framed<S>,
427    agreement: Agreement,
428}
429
430impl<S: Read + Write> Session<S> {
431    /// Connect on an explicitly advertised v2 endpoint. The caller sets a single
432    /// establishment deadline and verifies the peer before sending the offer.
433    pub fn connect(mut io: S, offer: &Hello) -> io::Result<Self> {
434        offer.validate()?;
435        io.write_all(&MAGIC)?;
436        let mut frames = Framed::new(io, HELLO_LIMIT)?;
437        frames.send(offer)?;
438        let remote: Hello = frames.receive()?;
439        check_refusal(&remote)?;
440        let agreement = negotiate(offer, &remote)?;
441        frames.io.write_all(&[1])?;
442        frames.io.flush()?;
443        read_acceptance(&mut frames.io)?;
444        frames.limit = agreement.max_frame;
445        Ok(Self { frames, agreement })
446    }
447
448    /// Admit a connection on the separate v2 listener, after authenticating its
449    /// OS peer. No request is dispatched if either peer rejects the negotiation.
450    pub fn accept(io: S, offer: &Hello) -> io::Result<Self> {
451        offer.validate()?;
452        let (mut frames, remote) = Self::offered(io)?;
453        let agreement = match negotiate(offer, &remote) {
454            Ok(agreement) => agreement,
455            Err(error) => {
456                if let Some(refusal) = rejection(offer, &remote) {
457                    // The failed negotiation is the error to report.
458                    let _ = frames.send(&refusal);
459                }
460                return Err(error);
461            }
462        };
463        frames.send(offer)?;
464        read_acceptance(&mut frames.io)?;
465        frames.io.write_all(&[1])?;
466        frames.io.flush()?;
467        frames.limit = agreement.max_frame;
468        Ok(Self { frames, agreement })
469    }
470
471    /// Read a client's offer and refuse it for `reason`, then close the
472    /// connection. Use it for a peer the policy rejected, or while retiring.
473    /// A client that does not read refusals, or that names another service,
474    /// sees only the close, as from an older server.
475    pub fn refuse(io: S, offer: &Hello, reason: Refused) -> io::Result<()> {
476        let (mut frames, remote) = Self::offered(io)?;
477        match reason.frame(offer, &remote) {
478            Some(refusal) => frames.send(&refusal),
479            None => Ok(()),
480        }
481    }
482
483    /// Read the magic and the client's offer.
484    fn offered(mut io: S) -> io::Result<(Framed<S>, Hello)> {
485        let mut magic = [0; 8];
486        io.read_exact(&mut magic)?;
487        if magic != MAGIC {
488            return Err(invalid("wrong binary protocol magic"));
489        }
490        let mut frames = Framed::new(io, HELLO_LIMIT)?;
491        let remote = frames.receive()?;
492        Ok((frames, remote))
493    }
494
495    /// Access the transport to change I/O deadlines after establishment.
496    /// Do not read or write protocol bytes through this reference.
497    pub fn transport_mut(&mut self) -> &mut S {
498        &mut self.frames.io
499    }
500
501    /// Negotiated version, capabilities and frame bound.
502    pub fn agreement(&self) -> Agreement {
503        self.agreement
504    }
505
506    /// Send a negotiated operation. An I/O failure has an uncertain outcome;
507    /// reconnecting must not replay it automatically.
508    pub fn send(&mut self, message: &Control) -> io::Result<()> {
509        self.check(message)?;
510        self.frames.send(message)
511    }
512
513    /// Receive a negotiated operation. Unknown operations terminate the session;
514    /// unknown optional fields within known operations remain compatible.
515    pub fn receive(&mut self) -> io::Result<Control> {
516        let message = self.frames.receive()?;
517        if let Err(error) = self.check(&message) {
518            self.frames.failed = true;
519            return Err(error);
520        }
521        Ok(message)
522    }
523
524    fn check(&self, message: &Control) -> io::Result<()> {
525        if required_capability(message)? & self.agreement.capabilities == 0 {
526            return Err(invalid("operation capability was not negotiated"));
527        }
528        Ok(())
529    }
530}
531
532fn read_acceptance(io: &mut impl Read) -> io::Result<()> {
533    let mut accepted = [0];
534    io.read_exact(&mut accepted)?;
535    if accepted != [1] {
536        return Err(invalid("peer did not accept negotiation"));
537    }
538    Ok(())
539}
540
541/// The client's endpoint choice is made before any socket bytes are sent.
542#[derive(Debug, PartialEq, Eq)]
543pub enum Endpoint<'a> {
544    /// Use the unchanged legacy protocol on this endpoint.
545    Legacy(&'a str),
546    /// Use binary negotiation exclusively on this advertised endpoint.
547    Binary(&'a str),
548}
549
550/// Prefer an explicitly advertised binary endpoint. An old discovery record
551/// selects legacy. Malformed v2 advertisements fail instead of downgrading.
552pub fn select_endpoint<'a>(legacy: &'a str, binary: Option<&'a str>) -> io::Result<Endpoint<'a>> {
553    if legacy.is_empty() {
554        return Err(invalid("missing legacy endpoint"));
555    }
556    match binary {
557        None => Ok(Endpoint::Legacy(legacy)),
558        Some(endpoint) if !endpoint.is_empty() && endpoint != legacy => {
559            Ok(Endpoint::Binary(endpoint))
560        }
561        Some(_) => Err(invalid("binary endpoint must be distinct and nonempty")),
562    }
563}
564
565#[cfg(feature = "wire-async")]
566mod asynchronous;
567#[cfg(feature = "wire-async")]
568pub use asynchronous::AsyncSession;
569
570/// Negotiation properties for every pair of valid offers.
571#[cfg(kani)]
572mod proofs {
573    use super::*;
574
575    fn valid_offer() -> Terms {
576        let terms: Terms = kani::any();
577        kani::assume(terms.valid());
578        terms
579    }
580
581    /// `Agreement` says both peers calculate it independently. They must reach
582    /// the same answer, or one side sends frames or operations the other
583    /// refuses.
584    #[kani::proof]
585    fn both_peers_reach_the_same_agreement() {
586        let a = valid_offer();
587        let b = valid_offer();
588        match (a.agree(b), b.agree(a)) {
589            (Ok(x), Ok(y)) => assert!(x == y),
590            (Err(_), Err(_)) => {}
591            _ => panic!("one peer agreed and the other refused"),
592        }
593    }
594
595    /// An agreement never exceeds what either peer offered, and every required
596    /// capability is part of it.
597    #[kani::proof]
598    fn an_agreement_stays_within_both_offers() {
599        let a = valid_offer();
600        let b = valid_offer();
601        if let Ok(agreed) = a.agree(b) {
602            assert!((MIN_FRAME..=MAX_FRAME).contains(&agreed.max_frame));
603            assert!(agreed.max_frame <= a.max_frame && agreed.max_frame <= b.max_frame);
604            assert!(agreed.capabilities & !a.supported == 0);
605            assert!(agreed.capabilities & !b.supported == 0);
606            assert!((a.required | b.required) & !agreed.capabilities == 0);
607            assert!(a.minimum <= agreed.version && agreed.version <= a.maximum);
608            assert!(b.minimum <= agreed.version && agreed.version <= b.maximum);
609        }
610    }
611}
kunobi_daemon/wire.rs

kunobi_daemon/
wire.rs
