kunobi_daemon/
client.rs

1//! Ask a running service for its health, or tell it to drain, over its binary
2//! control endpoint.
3//!
4//! [`health`] and [`drain`] check the kernel-reported peer PID against the
5//! reply and, when supplied, the expected PID. A PID match does not identify
6//! an executable or the author of a discovery record; see [`crate::peer`].
7//! The client also checks that the peer runs as this OS user and the reply
8//! arrives before the deadline.
9//!
10//! A failure after the service advertised its binary endpoint is final for
11//! that attempt: retry the same endpoint when [`RequestError::is_transient`]
12//! says so, or rediscover it, but never fall back to an older protocol.
13
14use crate::{
15    ProcessId,
16    local::{ConnectError, Duplex, Endpoint, PlatformDuplex, peer::PeerCredentials},
17    peer::{ExpectedProcess, Policy, Rejected, SameUser},
18    transport::SplitIo,
19    wire::{self, Control, Health, Hello, operation},
20};
21use std::{fmt, io, time::Instant};
22
23/// Why a control request did not produce a trusted reply.
24#[derive(Debug)]
25#[non_exhaustive]
26pub enum RequestError {
27    /// Nothing accepted the connection in time.
28    Unavailable(ConnectError),
29    /// The peer runs as another user or is not the expected process.
30    Peer(Rejected),
31    /// The OS could not read the peer credentials.
32    Credentials(io::Error),
33    /// The reply names a different process than the one on the connection.
34    ProcessMismatch {
35        /// The PID the kernel reports for the connection.
36        peer: ProcessId,
37        /// The PID the reply claims.
38        reported: u32,
39    },
40    /// The deadline passed before a reply arrived.
41    Late,
42    /// Negotiation, framing or the reply itself failed. A server that refused
43    /// the connection says why through [`wire::Refused::of`].
44    Protocol(io::Error),
45}
46
47impl RequestError {
48    /// Whether retrying the same endpoint can succeed: nothing was serving it
49    /// yet. Every other failure needs rediscovery or a different decision.
50    pub fn is_transient(&self) -> bool {
51        matches!(self, Self::Unavailable(error) if error.is_transient())
52    }
53}
54
55impl fmt::Display for RequestError {
56    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
57        match self {
58            Self::Unavailable(error) => write!(f, "control endpoint unavailable: {error}"),
59            Self::Peer(rejected) => write!(f, "control peer refused: {rejected}"),
60            Self::ProcessMismatch { peer, reported } => write!(
61                f,
62                "control reply names process {reported}, but the connection belongs to {peer}"
63            ),
64            Self::Late => f.write_str("control reply arrived after the deadline"),
65            Self::Credentials(error) => write!(f, "control peer credentials failed: {error}"),
66            Self::Protocol(error) => write!(f, "control exchange failed: {error}"),
67        }
68    }
69}
70
71impl std::error::Error for RequestError {
72    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
73        match self {
74            Self::Unavailable(error) => Some(error),
75            Self::Peer(rejected) => Some(rejected),
76            Self::Protocol(error) | Self::Credentials(error) => Some(error),
77            Self::ProcessMismatch { .. } | Self::Late => None,
78        }
79    }
80}
81
82/// Ask for health. See the module documentation.
83pub fn health(
84    endpoint: &Endpoint,
85    offer: &Hello,
86    expected: Option<ProcessId>,
87    deadline: Instant,
88) -> Result<Health, RequestError> {
89    request(endpoint, offer, operation::HEALTH, expected, deadline)
90}
91
92/// Close the service's admission. The reply reports the requests still active;
93/// it does not wait for them. See the module documentation.
94pub fn drain(
95    endpoint: &Endpoint,
96    offer: &Hello,
97    expected: Option<ProcessId>,
98    deadline: Instant,
99) -> Result<Health, RequestError> {
100    request(endpoint, offer, operation::DRAIN, expected, deadline)
101}
102
103/// Send one lifecycle `operation` to the binary control endpoint and return the
104/// reply, once the peer is this OS user, is `expected` when given, and the
105/// reply names that same process. Everything happens before `deadline`.
106fn request(
107    endpoint: &Endpoint,
108    offer: &Hello,
109    operation: u32,
110    expected: Option<ProcessId>,
111    deadline: Instant,
112) -> Result<Health, RequestError> {
113    let stream = PlatformDuplex::connect_once_until(endpoint, deadline)
114        .map_err(RequestError::Unavailable)?;
115    let evidence = stream.credentials().map_err(RequestError::Credentials)?;
116    match expected {
117        Some(pid) => ExpectedProcess::new(move || Some(pid))
118            .grant(&evidence)
119            .map(drop),
120        None => SameUser.grant(&evidence),
121    }
122    .map_err(RequestError::Peer)?;
123    let peer = evidence
124        .pid
125        .ok_or(RequestError::Peer(Rejected::Unreported))?;
126    let remaining = deadline.saturating_duration_since(Instant::now());
127    if remaining.is_zero() {
128        return Err(RequestError::Late);
129    }
130    stream
131        .set_read_deadline(Some(remaining))
132        .map_err(RequestError::Protocol)?;
133    let (read, write) = stream.split().map_err(RequestError::Protocol)?;
134    let mut session =
135        wire::Session::connect(SplitIo { read, write }, offer).map_err(RequestError::Protocol)?;
136    // One request per connection, so the reply needs no correlation ID.
137    let message = Control {
138        operation,
139        ..Default::default()
140    };
141    session.send(&message).map_err(RequestError::Protocol)?;
142    let reply = session.receive().map_err(RequestError::Protocol)?;
143    let health = Health::from_response(&reply, &message).map_err(RequestError::Protocol)?;
144    if health.process_id != peer.get() {
145        return Err(RequestError::ProcessMismatch {
146            peer,
147            reported: health.process_id,
148        });
149    }
150    if Instant::now() >= deadline {
151        return Err(RequestError::Late);
152    }
153    Ok(health)
154}
155
156#[cfg(test)]
157mod tests {
158    use super::*;
159
160    #[test]
161    fn only_an_endpoint_nothing_served_yet_is_worth_retrying() {
162        assert!(RequestError::Unavailable(ConnectError::ConnectTimeout).is_transient());
163        assert!(!RequestError::Unavailable(ConnectError::PermissionDenied).is_transient());
164        assert!(!RequestError::Unavailable(ConnectError::EndpointTooLong).is_transient());
165        assert!(!RequestError::Late.is_transient());
166        assert!(!RequestError::Peer(Rejected::OtherUser).is_transient());
167        assert!(!RequestError::Protocol(io::ErrorKind::Other.into()).is_transient());
168    }
169
170    #[test]
171    fn nested_control_errors_keep_their_sources() {
172        for error in [
173            RequestError::Unavailable(ConnectError::Failed(io::ErrorKind::InvalidInput)),
174            RequestError::Peer(Rejected::OtherUser),
175            RequestError::Credentials(io::Error::other("credential query denied")),
176            RequestError::Protocol(io::Error::other("bad frame")),
177        ] {
178            assert!(std::error::Error::source(&error).is_some());
179            assert!(!error.is_transient());
180        }
181        assert!(
182            std::error::Error::source(&RequestError::ProcessMismatch {
183                peer: ProcessId::new(7).unwrap(),
184                reported: 9,
185            })
186            .is_none()
187        );
188    }
189
190    #[test]
191    fn every_failure_says_what_happened() {
192        let peer = ProcessId::new(7).unwrap();
193        let cases: [(RequestError, &str); 6] = [
194            (
195                RequestError::Credentials(io::Error::other("denied")),
196                "control peer credentials failed: denied",
197            ),
198            (
199                RequestError::Unavailable(ConnectError::ConnectTimeout),
200                "control endpoint unavailable: daemon connection timed out",
201            ),
202            (
203                RequestError::Peer(Rejected::OtherUser),
204                "control peer refused: the peer runs as another OS user",
205            ),
206            (
207                RequestError::ProcessMismatch { peer, reported: 9 },
208                "control reply names process 9, but the connection belongs to 7",
209            ),
210            (
211                RequestError::Late,
212                "control reply arrived after the deadline",
213            ),
214            (
215                RequestError::Protocol(io::Error::other("bad frame")),
216                "control exchange failed: bad frame",
217            ),
218        ];
219        for (error, text) in cases {
220            assert_eq!(error.to_string(), text);
221        }
222        assert!(std::error::Error::source(&RequestError::Late).is_none());
223        assert!(std::error::Error::source(&RequestError::Peer(Rejected::OtherUser)).is_some());
224    }
225}
kunobi_daemon/client.rs

kunobi_daemon/
client.rs
