mkit_cli/grants/
remote.rs

1//! Picking the remote a grant, epoch or visibility command talks to, the
2//! audience it signs for, and waiting for the server to finish (WP-2.14).
3
4use std::time::Duration;
5
6use mkit_attest::grant::{Namespace, is_loopback_origin};
7use mkit_core::repo_identity::RepositoryIdentity;
8use mkit_transport_connect::{Completion, audience_from_url, repository_identity_from_url};
9
10use crate::config::{self, LayeredConfig};
11
12/// A Connect remote resolved from a remote name or an `mkit+https://` URL.
13#[derive(Debug, Clone, PartialEq, Eq)]
14pub struct Target {
15    pub name: String,
16    pub endpoint: String,
17    /// The repository config chose it, so the credential-trust gate applies.
18    pub repo_chosen: bool,
19}
20
21impl Target {
22    /// The auth v2 audience for this remote: its origin.
23    ///
24    /// # Errors
25    /// The endpoint is not an `mkit+https://` or loopback `mkit+http://` URL.
26    pub fn audience(&self) -> Result<String, String> {
27        audience_from_url(&self.endpoint).ok_or_else(|| {
28            format!(
29                "`{}` is not an mkit+https:// (or loopback mkit+http://) remote",
30                self.endpoint
31            )
32        })
33    }
34
35    /// The repository the URL names, if it names a full `<namespace>/<name>`.
36    #[must_use]
37    pub fn repository(&self) -> Option<RepositoryIdentity> {
38        repository_identity_from_url(&self.endpoint)
39            .ok()
40            .filter(|id| id.namespace().is_some())
41    }
42
43    /// The namespace in the URL's repository, if any.
44    #[must_use]
45    pub fn namespace(&self) -> Option<Namespace> {
46        self.repository().and_then(|id| id.namespace().copied())
47    }
48
49    /// Whether this is a loopback `mkit+http://` remote: the one place a
50    /// loopback audience is acceptable (a development server).
51    #[must_use]
52    pub fn is_loopback_dev(&self) -> bool {
53        self.endpoint.starts_with("mkit+http://")
54            && self.audience().is_ok_and(|a| is_loopback_origin(&a))
55    }
56}
57
58/// Resolve `arg` as a remote name, or take it as a URL if it starts with
59/// `mkit+`. With no `arg`, the user's `trusted_remote_endpoint`.
60///
61/// # Errors
62/// An unknown remote, or nothing to fall back on.
63pub fn resolve_target(cfg: &LayeredConfig, arg: Option<&str>) -> Result<Target, String> {
64    match arg {
65        Some(url) if url.starts_with("mkit+") => Ok(Target {
66            name: url.to_owned(),
67            endpoint: url.to_owned(),
68            repo_chosen: false,
69        }),
70        Some(name) => {
71            let resolved = config::resolve_remote(cfg, name).ok_or_else(|| {
72                format!("no remote named `{name}` (add one with `mkit remote add`, or pass an mkit+https:// URL)")
73            })?;
74            Ok(Target {
75                name: resolved.name,
76                endpoint: resolved.endpoint,
77                repo_chosen: resolved.repo_chosen,
78            })
79        }
80        None => {
81            let trusted = cfg.user.trusted_remote_endpoint.trim();
82            if trusted.is_empty() {
83                return Err(
84                    "no remote given and no trusted remote configured (set one with `mkit config trusted_remote_endpoint <url>`)"
85                        .to_owned(),
86                );
87            }
88            Ok(Target {
89                name: "trusted".to_owned(),
90                endpoint: trusted.to_owned(),
91                repo_chosen: false,
92            })
93        }
94    }
95}
96
97/// SPEC-WRITE-GRANTS §3.2: a loopback audience is refused unless the remote
98/// itself is a loopback `mkit+http://` development remote. Every local
99/// deployment shares a loopback audience, so a statement signed for one
100/// would verify at all of them.
101///
102/// # Errors
103/// A loopback audience and a remote that isn't a loopback dev remote.
104pub fn check_audiences(audiences: &[String], remote: Option<&Target>) -> Result<(), String> {
105    for audience in audiences {
106        if is_loopback_origin(audience) && !remote.is_some_and(Target::is_loopback_dev) {
107            return Err(format!(
108                "audience {audience} is a loopback address, which every local server shares; \
109                 refusing to sign for it unless the remote is itself a loopback mkit+http:// development remote"
110            ));
111        }
112    }
113    Ok(())
114}
115
116/// How [`drive`] ended.
117#[derive(Debug, PartialEq, Eq)]
118pub enum Driven<T> {
119    Done(T),
120    /// The next wait would pass `timeout`; `waited` was spent.
121    TimedOut {
122        waited: Duration,
123    },
124    /// `sleep` returned false (Ctrl-C).
125    Cancelled,
126}
127
128/// Call `call` until it is [`Completion::Done`], waiting the server's
129/// `Retry-After` between calls. `call` sends the same request every time: the
130/// caller signs once and re-sends the identical statement bytes (SPEC-WRITE-GRANTS
131/// §5.2: the same epoch is a retry), never a fresh nonce.
132///
133/// `timeout` is a wall-clock bound measured from the first call: a re-send is
134/// not started when the time already spent plus the next delay would pass it.
135/// `sleep` may cancel (Ctrl-C). Cancellation is observed while waiting, not
136/// inside an RPC in flight (each RPC is bounded by the transport's own timeout).
137///
138/// # Errors
139/// Whatever `call` returns.
140pub fn drive<T, E>(
141    call: impl FnMut() -> Result<Completion<T>, E>,
142    timeout: Duration,
143    sleep: impl FnMut(Duration) -> bool,
144) -> Result<Driven<T>, E> {
145    let start = std::time::Instant::now();
146    drive_with_clock(call, timeout, sleep, || start.elapsed())
147}
148
149fn drive_with_clock<T, E>(
150    mut call: impl FnMut() -> Result<Completion<T>, E>,
151    timeout: Duration,
152    mut sleep: impl FnMut(Duration) -> bool,
153    elapsed: impl Fn() -> Duration,
154) -> Result<Driven<T>, E> {
155    loop {
156        match call()? {
157            Completion::Done(value) => return Ok(Driven::Done(value)),
158            Completion::Pending { retry_after } => {
159                let waited = elapsed();
160                if waited.saturating_add(retry_after) > timeout {
161                    return Ok(Driven::TimedOut { waited });
162                }
163                if !sleep(retry_after) {
164                    return Ok(Driven::Cancelled);
165                }
166            }
167        }
168    }
169}
170
171/// Sleep for `duration` in short slices so Ctrl-C is prompt. Returns false if
172/// the process was asked to shut down.
173#[must_use]
174pub fn interruptible_sleep(duration: Duration) -> bool {
175    let slice = Duration::from_millis(100);
176    let mut left = duration;
177    while !left.is_zero() {
178        if crate::signal::is_shutdown() {
179            return false;
180        }
181        let step = left.min(slice);
182        std::thread::sleep(step);
183        left -= step;
184    }
185    !crate::signal::is_shutdown()
186}
187
188#[cfg(test)]
189mod tests {
190    use super::*;
191
192    fn target(url: &str) -> Target {
193        Target {
194            name: "t".into(),
195            endpoint: url.into(),
196            repo_chosen: false,
197        }
198    }
199
200    #[test]
201    fn loopback_audiences_need_a_loopback_dev_remote() {
202        let audiences = |a: &str| vec![a.to_owned()];
203        let dev = target("mkit+http://127.0.0.1:8080");
204        let prod = target("mkit+https://git.example.com");
205        for loopback in [
206            "http://localhost:8080",
207            "http://127.0.0.1:8080",
208            "http://127.9.9.9",
209            "http://[::1]:8080",
210        ] {
211            assert!(
212                check_audiences(&audiences(loopback), Some(&prod)).is_err(),
213                "{loopback}"
214            );
215            assert!(
216                check_audiences(&audiences(loopback), None).is_err(),
217                "{loopback}"
218            );
219        }
220        assert!(check_audiences(&audiences("http://127.0.0.1:8080"), Some(&dev)).is_ok());
221        assert!(check_audiences(&audiences("https://git.example.com"), Some(&prod)).is_ok());
222        assert!(target("mkit+http://127.0.0.1:8080").is_loopback_dev());
223        assert!(!target("mkit+https://git.example.com").is_loopback_dev());
224    }
225
226    #[test]
227    fn audience_and_namespace_come_from_the_url() {
228        let t =
229            target("mkit+https://git.example.com/0x8ba1f109551bd432803012645ac136ddd64dba72/site");
230        assert_eq!(t.audience().unwrap(), "https://git.example.com");
231        assert_eq!(
232            t.namespace().unwrap().to_string(),
233            "0x8ba1f109551bd432803012645ac136ddd64dba72"
234        );
235        assert!(target("mkit+https://git.example.com").namespace().is_none());
236        assert!(target("https://git.example.com").audience().is_err());
237    }
238
239    #[test]
240    fn drive_resends_until_done_and_sums_the_waits() {
241        let mut calls: u64 = 0;
242        let mut slept = Vec::new();
243        let clock = std::cell::Cell::new(Duration::ZERO);
244        let outcome: Result<Driven<u64>, ()> = drive_with_clock(
245            || {
246                calls += 1;
247                Ok(if calls < 4 {
248                    Completion::Pending {
249                        retry_after: Duration::from_secs(calls),
250                    }
251                } else {
252                    Completion::Done(9)
253                })
254            },
255            Duration::from_mins(1),
256            |d| {
257                slept.push(d);
258                clock.set(clock.get() + d);
259                true
260            },
261            || clock.get(),
262        );
263        assert_eq!(outcome, Ok(Driven::Done(9)));
264        assert_eq!(calls, 4);
265        assert_eq!(
266            slept,
267            [
268                Duration::from_secs(1),
269                Duration::from_secs(2),
270                Duration::from_secs(3)
271            ]
272        );
273    }
274
275    #[test]
276    fn drive_stops_at_the_total_timeout_and_on_cancel() {
277        let pending = || -> Result<Completion<()>, ()> {
278            Ok(Completion::Pending {
279                retry_after: Duration::from_secs(4),
280            })
281        };
282        // 4 + 4 fit in 10 s; a third wait would end at 12 s.
283        let mut sleeps = 0;
284        let clock = std::cell::Cell::new(Duration::ZERO);
285        let outcome = drive_with_clock(
286            pending,
287            Duration::from_secs(10),
288            |d| {
289                sleeps += 1;
290                clock.set(clock.get() + d);
291                true
292            },
293            || clock.get(),
294        );
295        assert_eq!(
296            outcome,
297            Ok(Driven::TimedOut {
298                waited: Duration::from_secs(8)
299            })
300        );
301        assert_eq!(sleeps, 2);
302        let outcome = drive(pending, Duration::from_secs(10), |_| false);
303        assert_eq!(outcome, Ok(Driven::Cancelled));
304        // Time spent inside the calls counts: 7 s per call leaves no room for
305        // a 4 s wait inside a 10 s bound.
306        let clock = std::cell::Cell::new(Duration::ZERO);
307        let outcome = drive_with_clock(
308            || {
309                clock.set(clock.get() + Duration::from_secs(7));
310                pending()
311            },
312            Duration::from_secs(10),
313            |_| true,
314            || clock.get(),
315        );
316        assert_eq!(
317            outcome,
318            Ok(Driven::TimedOut {
319                waited: Duration::from_secs(7)
320            })
321        );
322        // Errors from the call pass straight through.
323        let outcome: Result<Driven<()>, &str> =
324            drive(|| Err("boom"), Duration::from_secs(1), |_| true);
325        assert_eq!(outcome, Err("boom"));
326    }
327}
mkit_cli/grants/remote.rs

mkit_cli/grants/
remote.rs
