kernoid_client/
lib.rs

1// SPDX-License-Identifier: MIT
2//! kernoid's Rust SDK for clients. JMAP (RFC 8620) over HTTPS: the session, method calls with result references,
3//! uploads, downloads and the push stream. Typed calls for mail, files, calendars, contacts and principals are in the
4//! submodules; end-to-end encrypted mail in `e2e` (RFC 0006), finding a server in `srv`, iCalendar and vCard files in
5//! `transfer`. Everything is async on tokio; the terminal app and the apps share it.
6
7pub mod account;
8pub mod calendars;
9pub mod contacts;
10pub mod e2e;
11pub mod files;
12pub mod mail;
13pub mod oidc;
14pub mod push;
15pub mod reply;
16pub mod srv;
17pub mod transfer;
18
19use std::{
20    collections::BTreeMap,
21    fmt,
22    net::{IpAddr, Ipv4Addr, SocketAddr},
23    time::Duration,
24};
25
26use base64::Engine;
27use futures_util::{Stream, StreamExt, TryStreamExt};
28use serde::Deserialize;
29use serde_json::{Map, Value, json};
30use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt};
31
32pub const CORE: &str = "urn:ietf:params:jmap:core";
33pub const MAIL: &str = "urn:ietf:params:jmap:mail";
34pub const SUBMISSION: &str = "urn:ietf:params:jmap:submission";
35pub const CALENDARS: &str = "urn:ietf:params:jmap:calendars";
36pub const CONTACTS: &str = "urn:ietf:params:jmap:contacts";
37pub const FILENODE: &str = "urn:ietf:params:jmap:filenode";
38/// kernoid's extension for the versions of a file (`FileVersion/get`).
39pub const FILE_VERSIONS: &str = "https://kernoid.dev/jmap/file-versions";
40/// kernoid's extension for links to files for people without an account (`FileLink/*`).
41pub const FILE_LINKS: &str = "https://kernoid.dev/jmap/file-links";
42pub const PRINCIPALS: &str = "urn:ietf:params:jmap:principals";
43pub const VACATION: &str = "urn:ietf:params:jmap:vacationresponse";
44
45/// What went wrong talking to the server.
46#[derive(Debug)]
47pub enum Error {
48    /// The address and password were not accepted (401).
49    Unauthorized,
50    /// The server answered with another status than expected; with the detail of its problem, if it sent one.
51    Status(u16, Option<String>),
52    /// The connection failed or broke.
53    Transport(String),
54    /// A method call failed (RFC 8620, 3.6.2) or a request was refused as a whole (3.6.1).
55    Method { kind: String, description: Option<String> },
56    /// A change of `*/set` was refused (RFC 8620, 5.3).
57    Set { kind: String, description: Option<String> },
58    /// The answer was not what JMAP says it is.
59    Protocol(String),
60    /// Reading or writing a local file.
61    Io(std::io::Error),
62    /// End-to-end encryption: unlocking the key, encrypting, decrypting (RFC 0006).
63    E2e(String),
64}
65
66impl fmt::Display for Error {
67    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
68        match self {
69            Self::Unauthorized => write!(f, "the address or password was not accepted"),
70            Self::Status(status, Some(detail)) => write!(f, "the server answered {status}: {detail}"),
71            Self::Status(status, None) => write!(f, "the server answered {status}"),
72            Self::Transport(error) => write!(f, "connection: {error}"),
73            Self::Method { kind, description: Some(description) }
74            | Self::Set { kind, description: Some(description) } => {
75                write!(f, "{kind}: {description}")
76            }
77            Self::Method { kind, description: None } | Self::Set { kind, description: None } => write!(f, "{kind}"),
78            Self::Protocol(error) => write!(f, "unexpected answer: {error}"),
79            Self::Io(error) => write!(f, "{error}"),
80            Self::E2e(error) => write!(f, "end-to-end encryption: {error}"),
81        }
82    }
83}
84
85impl std::error::Error for Error {}
86
87impl From<reqwest::Error> for Error {
88    fn from(error: reqwest::Error) -> Self {
89        Self::Transport(error.to_string())
90    }
91}
92
93impl From<kernoid_e2e::Error> for Error {
94    fn from(error: kernoid_e2e::Error) -> Self {
95        Self::E2e(error.to_string())
96    }
97}
98
99impl From<std::io::Error> for Error {
100    fn from(error: std::io::Error) -> Self {
101        Self::Io(error)
102    }
103}
104
105pub type Result<T> = std::result::Result<T, Error>;
106
107/// An account in the session (RFC 8620, 2): the user's own, one shared with them, or a drive of their tenant.
108#[derive(Debug, Clone, Deserialize)]
109#[serde(rename_all = "camelCase")]
110pub struct Account {
111    pub name: String,
112    #[serde(default)]
113    pub is_personal: bool,
114    #[serde(default)]
115    pub is_read_only: bool,
116    #[serde(default)]
117    pub account_capabilities: Map<String, Value>,
118    /// kernoid marks the accounts that are drives – folders the tenant owns (RFC 0005).
119    #[serde(default, rename = "kernoid:drive")]
120    pub drive: bool,
121}
122
123/// The session resource (RFC 8620, 2).
124#[derive(Debug, Clone, Deserialize)]
125#[serde(rename_all = "camelCase")]
126pub struct Session {
127    #[serde(default)]
128    pub capabilities: Map<String, Value>,
129    pub accounts: BTreeMap<String, Account>,
130    #[serde(default)]
131    pub primary_accounts: BTreeMap<String, String>,
132    pub username: String,
133    pub api_url: String,
134    pub download_url: String,
135    pub upload_url: String,
136    #[serde(default)]
137    pub event_source_url: String,
138    #[serde(default)]
139    pub state: String,
140}
141
142impl Session {
143    /// The user's own account for a capability.
144    pub fn primary(&self, capability: &str) -> Option<&str> {
145        self.primary_accounts.get(capability).map(String::as_str)
146    }
147
148    /// The accounts that have a capability, the primary one first, then by name.
149    pub fn accounts_with(&self, capability: &str) -> Vec<(&str, &Account)> {
150        let primary = self.primary(capability);
151        let mut out: Vec<_> = self
152            .accounts
153            .iter()
154            .filter(|(_, account)| account.account_capabilities.contains_key(capability))
155            .map(|(id, account)| (id.as_str(), account))
156            .collect();
157        out.sort_by(|a, b| (Some(a.0) != primary, &a.1.name).cmp(&(Some(b.0) != primary, &b.1.name)));
158        out
159    }
160}
161
162/// The responses to one request, in order: name, arguments, call id.
163#[derive(Debug, Clone)]
164pub struct Responses(pub Vec<(String, Value, String)>);
165
166impl Responses {
167    /// The (first) response to a call id.
168    pub fn get(&self, id: &str) -> Option<&Value> {
169        self.0.iter().find(|(_, _, call)| call == id).map(|(_, value, _)| value)
170    }
171
172    /// The (first) response to a call id, as `T`.
173    pub fn parse<T: serde::de::DeserializeOwned>(&self, id: &str) -> Result<T> {
174        let value = self.get(id).ok_or_else(|| Error::Protocol(format!("no response to {id}")))?;
175        serde_json::from_value(value.clone()).map_err(|error| Error::Protocol(format!("{id}: {error}")))
176    }
177}
178
179/// The scheme and authority of an absolute URL (`https://host:port`), and the rest (path and query, at least `/`).
180pub fn split_origin(url: &str) -> Option<(&str, &str)> {
181    let (scheme, after) = url.split_once("://")?;
182    if scheme.is_empty() || !scheme.bytes().all(|b| b.is_ascii_alphanumeric() || b"+-.".contains(&b)) {
183        return None;
184    }
185    let end = after.find(['/', '?', '#']).unwrap_or(after.len());
186    if end == 0 {
187        return None;
188    }
189    let origin = &url[..scheme.len() + 3 + end];
190    let rest = &after[end..];
191    Some((origin, if rest.is_empty() { "/" } else { rest }))
192}
193
194/// The host of an origin, without port and brackets.
195fn host(origin: &str) -> &str {
196    let authority = origin.split_once("://").map_or(origin, |(_, rest)| rest);
197    let authority = authority.rsplit_once('@').map_or(authority, |(_, host)| host);
198    if let Some(rest) = authority.strip_prefix('[') {
199        return rest.split_once(']').map_or(rest, |(host, _)| host);
200    }
201    authority.split_once(':').map_or(authority, |(host, _)| host)
202}
203
204/// Whether plain HTTP to this origin stays on this machine: loopback addresses, `localhost` and `*.localhost`.
205pub fn is_local(origin: &str) -> bool {
206    let host = host(origin).to_ascii_lowercase();
207    host == "localhost"
208        || host.ends_with(".localhost")
209        || host.parse::<std::net::IpAddr>().is_ok_and(|address| address.is_loopback())
210}
211
212/// Percent-encodes everything but RFC 3986's unreserved characters, for the variables of URL templates.
213pub fn encode(value: &str) -> String {
214    let mut out = String::with_capacity(value.len());
215    for byte in value.bytes() {
216        if byte.is_ascii_alphanumeric() || b"-._~".contains(&byte) {
217            out.push(byte as char);
218        } else {
219            out.push_str(&format!("%{byte:02X}"));
220        }
221    }
222    out
223}
224
225/// Where to find the session for what the user gave: a server (`https://host`, with or without a path) or, without
226/// one, the domain of the address – `https://<domain>/.well-known/jmap` (RFC 8620, 2.2).
227pub fn session_url(server: Option<&str>, address: &str) -> Result<String> {
228    let server = match server {
229        Some(server) if server.contains("://") => server.trim_end_matches('/').to_owned(),
230        Some(server) => format!("https://{}", server.trim_end_matches('/')),
231        None => {
232            let domain = address
233                .rsplit_once('@')
234                .map(|(_, domain)| domain)
235                .filter(|domain| !domain.is_empty())
236                .ok_or_else(|| Error::Protocol("give a server, or an address with a domain".into()))?;
237            format!("https://{domain}")
238        }
239    };
240    let (_, path) = split_origin(&server).ok_or_else(|| Error::Protocol(format!("not a URL: {server}")))?;
241    Ok(if path == "/" { format!("{server}/.well-known/jmap") } else { server })
242}
243
244/// How a connection is made.
245#[derive(Debug, Clone)]
246pub struct Options {
247    /// Allow plain HTTP to hosts other than this machine (the password would travel readable).
248    pub allow_http: bool,
249    /// Who is asking, in `User-Agent` (`kernoid-tui/0.2.3`).
250    pub user_agent: String,
251}
252
253impl Default for Options {
254    fn default() -> Self {
255        Self { allow_http: false, user_agent: concat!("kernoid-client/", env!("CARGO_PKG_VERSION")).into() }
256    }
257}
258
259/// What a request to kernoid's own API carries.
260pub(crate) enum Body {
261    Json(Value),
262    Bytes(Vec<u8>, String),
263}
264
265/// A logged-in JMAP client: the session and the credentials, sent only to the origin the session came from.
266#[derive(Clone)]
267pub struct Client {
268    http: reqwest::Client,
269    /// A client without overall timeout, for transfers and the push stream.
270    streaming: reqwest::Client,
271    /// `Basic …` or `Bearer …`; shared by the clones, so that a renewed access token reaches them all.
272    authorization: std::sync::Arc<std::sync::RwLock<String>>,
273    origin: String,
274    session_url: String,
275    pub session: Session,
276}
277
278impl fmt::Debug for Client {
279    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
280        f.debug_struct("Client").field("origin", &self.origin).field("username", &self.session.username).finish()
281    }
282}
283
284/// The problem detail of an error answer (RFC 9457), if it has one.
285fn problem_detail(body: &str) -> Option<String> {
286    let value: Value = serde_json::from_str(body).ok()?;
287    value.get("detail").or_else(|| value.get("description")).and_then(Value::as_str).map(str::to_owned)
288}
289
290/// An HTTP client: TLS through reqwest's rustls on AWS-LC with the system's trust store (rustls-platform-verifier),
291/// post-quantum key exchange preferred (the feature `prefer-post-quantum` of `rustls`); redirects followed up to five
292/// times – reqwest drops the credentials on a redirect to another host.
293fn http(origin: &str, user_agent: &str, timeout: Option<Duration>) -> Result<reqwest::Client> {
294    let mut builder = reqwest::Client::builder()
295        .redirect(reqwest::redirect::Policy::limited(5))
296        .connect_timeout(Duration::from_secs(15))
297        .user_agent(user_agent);
298    if let Some(timeout) = timeout {
299        builder = builder.timeout(timeout);
300    }
301    // `*.localhost` is this machine (RFC 6761), but not every system's resolver knows it.
302    let name = host(origin);
303    if name.ends_with(".localhost") {
304        builder = builder.resolve(name, SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0));
305    }
306    Ok(builder.build()?)
307}
308
309impl Client {
310    /// Fetches the session at `session_url` with HTTP Basic – the address and an app password.
311    pub async fn connect(session_url: &str, username: &str, password: &str, options: &Options) -> Result<Self> {
312        let credentials = base64::engine::general_purpose::STANDARD.encode(format!("{username}:{password}"));
313        Self::open(session_url, username, format!("Basic {credentials}"), options).await
314    }
315
316    /// Fetches the session at `session_url` with an OpenID Connect access token of kernoid with the scope
317    /// `urn:kernoid:data` – from [`oidc`]. Renew it with [`Client::set_access_token`] before it expires.
318    pub async fn connect_with_token(session_url: &str, access_token: &str, options: &Options) -> Result<Self> {
319        Self::open(session_url, "", format!("Bearer {access_token}"), options).await
320    }
321
322    /// A new access token for this client and its clones (after [`oidc::refresh`]).
323    pub fn set_access_token(&self, access_token: &str) {
324        *self.authorization.write().unwrap_or_else(|poisoned| poisoned.into_inner()) = format!("Bearer {access_token}");
325    }
326
327    fn authorization(&self) -> String {
328        self.authorization.read().unwrap_or_else(|poisoned| poisoned.into_inner()).clone()
329    }
330
331    async fn open(session_url: &str, username: &str, authorization: String, options: &Options) -> Result<Self> {
332        let (origin, _) =
333            split_origin(session_url).ok_or_else(|| Error::Protocol(format!("not a URL: {session_url}")))?;
334        let origin = origin.to_owned();
335        if !origin.starts_with("https://")
336            && !(options.allow_http && origin.starts_with("http://"))
337            && !is_local(&origin)
338        {
339            return Err(Error::Protocol(format!("{origin} is not HTTPS; the password would travel readable")));
340        }
341        let mut client = Self {
342            http: http(&origin, &options.user_agent, Some(Duration::from_secs(120)))?,
343            streaming: http(&origin, &options.user_agent, None)?,
344            authorization: std::sync::Arc::new(std::sync::RwLock::new(authorization)),
345            origin,
346            session_url: session_url.to_owned(),
347            session: Session {
348                capabilities: Map::new(),
349                accounts: BTreeMap::new(),
350                primary_accounts: BTreeMap::new(),
351                username: username.to_owned(),
352                api_url: String::new(),
353                download_url: String::new(),
354                upload_url: String::new(),
355                event_source_url: String::new(),
356                state: String::new(),
357            },
358        };
359        client.session = client.fetch_session(session_url).await?;
360        Ok(client)
361    }
362
363    /// Fetches the session again (after a `StateChange` of the session, or to see new accounts).
364    pub async fn refresh_session(&mut self) -> Result<()> {
365        self.session = self.fetch_session(&self.session_url.clone()).await?;
366        Ok(())
367    }
368
369    async fn fetch_session(&self, url: &str) -> Result<Session> {
370        let response = self
371            .http
372            .get(url)
373            .header("authorization", self.authorization())
374            .header("accept", "application/json")
375            .send()
376            .await?;
377        let body = check(response, 200).await?;
378        serde_json::from_str(&body).map_err(|error| Error::Protocol(format!("the session: {error}")))
379    }
380
381    /// The origin this client talks to.
382    pub fn origin(&self) -> &str {
383        &self.origin
384    }
385
386    /// A URL of the session on the origin the session came from – the credentials go nowhere else. A server that
387    /// serves its API on another host is not supported (OPEN).
388    fn here(&self, url: &str) -> String {
389        match split_origin(url) {
390            Some((_, rest)) => format!("{}{rest}", self.origin),
391            None if url.starts_with('/') => format!("{}{url}", self.origin),
392            None => format!("{}/{url}", self.origin),
393        }
394    }
395
396    /// The user's own account for a capability.
397    pub fn primary(&self, capability: &str) -> Result<String> {
398        self.session
399            .primary(capability)
400            .map(str::to_owned)
401            .ok_or_else(|| Error::Method { kind: "accountNotFound".into(), description: Some(capability.into()) })
402    }
403
404    /// Runs method calls; the first method error fails them all.
405    pub async fn request(&self, using: &[&str], calls: Vec<(&str, Value, &str)>) -> Result<Responses> {
406        let mut capabilities = vec![CORE];
407        capabilities.extend(using.iter().filter(|capability| **capability != CORE));
408        let calls: Vec<Value> = calls.into_iter().map(|(name, arguments, id)| json!([name, arguments, id])).collect();
409        let body = json!({ "using": capabilities, "methodCalls": calls }).to_string();
410        let response = self
411            .http
412            .post(self.here(&self.session.api_url))
413            .header("authorization", self.authorization())
414            .header("content-type", "application/json")
415            .header("accept", "application/json")
416            .body(body)
417            .send()
418            .await?;
419        let status = response.status().as_u16();
420        let text = limited_text(response, 64 << 20).await?;
421        if status == 401 {
422            return Err(Error::Unauthorized);
423        }
424        let value: Value = serde_json::from_str(&text).map_err(|_| Error::Status(status, problem_detail(&text)))?;
425        if status != 200 {
426            // A request refused as a whole (RFC 8620, 3.6.1) is a problem with a type.
427            let kind = value.get("type").and_then(Value::as_str).unwrap_or("serverFail");
428            let kind = kind.rsplit(':').next().unwrap_or(kind).to_owned();
429            return Err(Error::Method { kind, description: problem_detail(&text) });
430        }
431        let responses = value
432            .get("methodResponses")
433            .and_then(Value::as_array)
434            .ok_or_else(|| Error::Protocol("no methodResponses".into()))?;
435        let mut out = Vec::with_capacity(responses.len());
436        for response in responses {
437            let (Some(name), Some(arguments), Some(id)) =
438                (response.get(0).and_then(Value::as_str), response.get(1), response.get(2).and_then(Value::as_str))
439            else {
440                return Err(Error::Protocol("a malformed method response".into()));
441            };
442            if name == "error" {
443                return Err(Error::Method {
444                    kind: arguments.get("type").and_then(Value::as_str).unwrap_or("serverFail").to_owned(),
445                    description: arguments.get("description").and_then(Value::as_str).map(str::to_owned),
446                });
447            }
448            out.push((name.to_owned(), arguments.clone(), id.to_owned()));
449        }
450        Ok(Responses(out))
451    }
452
453    /// One method call; its response.
454    pub async fn call(&self, using: &[&str], name: &str, arguments: Value) -> Result<Value> {
455        let responses = self.request(using, vec![(name, arguments, "0")]).await?;
456        responses.get("0").cloned().ok_or_else(|| Error::Protocol(format!("no response to {name}")))
457    }
458
459    /// A request to kernoid's own API (`/v1/…`, Problem Details on errors) on the origin of the session, with the
460    /// login's credentials; the answer's body, empty for `204`. `body` is JSON, or bytes with their media type.
461    pub(crate) async fn rest(&self, method: reqwest::Method, path: &str, body: Option<Body>) -> Result<String> {
462        let mut request = self
463            .http
464            .request(method, format!("{}{path}", self.origin))
465            .header("authorization", self.authorization())
466            .header("accept", "application/json");
467        request = match body {
468            Some(Body::Json(value)) => request.header("content-type", "application/json").body(value.to_string()),
469            Some(Body::Bytes(bytes, kind)) => request.header("content-type", kind).body(bytes),
470            None => request,
471        };
472        let response = request.send().await?;
473        let status = response.status().as_u16();
474        let text = limited_text(response, 4 << 20).await?;
475        match status {
476            401 => Err(Error::Unauthorized),
477            200..=299 => Ok(text),
478            status => Err(Error::Status(status, problem_detail(&text))),
479        }
480    }
481
482    /// Bytes from kernoid's own API – a picture.
483    pub(crate) async fn rest_bytes(&self, path: &str) -> Result<Vec<u8>> {
484        let response = self
485            .http
486            .get(format!("{}{path}", self.origin))
487            .header("authorization", self.authorization())
488            .send()
489            .await?;
490        match response.status().as_u16() {
491            401 => Err(Error::Unauthorized),
492            200 => Ok(response.bytes().await?.to_vec()),
493            status => Err(Error::Status(status, None)),
494        }
495    }
496
497    /// `<type>/set` (RFC 8620, 5.3); the first refused change fails it. `extra` adds arguments
498    /// (`onDestroyRemoveChildren`, `onExists`).
499    pub async fn set(&self, using: &[&str], kind: &str, account: &str, change: Value, extra: Value) -> Result<Value> {
500        let mut arguments = json!({ "accountId": account });
501        for source in [change, extra] {
502            if let (Some(target), Value::Object(source)) = (arguments.as_object_mut(), source) {
503                target.extend(source);
504            }
505        }
506        let result = self.call(using, &format!("{kind}/set"), arguments).await?;
507        set_error(&result)?;
508        Ok(result)
509    }
510
511    /// The URL a blob is downloaded from (RFC 8620, 6.2).
512    pub fn download_url(&self, account: &str, blob: &str, name: &str, kind: &str) -> String {
513        let url = self
514            .session
515            .download_url
516            .replace("{accountId}", &encode(account))
517            .replace("{blobId}", &encode(blob))
518            .replace("{name}", &encode(name))
519            .replace("{type}", &encode(kind));
520        self.here(&url)
521    }
522
523    /// Downloads a blob into `out`; `progress` hears the bytes so far and the whole, if the server said it.
524    pub async fn download(
525        &self,
526        account: &str,
527        blob: &str,
528        name: &str,
529        kind: &str,
530        out: &mut (dyn AsyncWrite + Unpin + Send),
531        progress: &mut (dyn FnMut(u64, Option<u64>) + Send),
532    ) -> Result<u64> {
533        let url = self.download_url(account, blob, name, kind);
534        let response = self.streaming.get(url).header("authorization", self.authorization()).send().await?;
535        let status = response.status().as_u16();
536        if status == 401 {
537            return Err(Error::Unauthorized);
538        }
539        if status != 200 {
540            let body = limited_text(response, 16 << 20).await.unwrap_or_default();
541            return Err(Error::Status(status, problem_detail(&body)));
542        }
543        let total = response.content_length();
544        let mut body = response.bytes_stream();
545        let mut done = 0;
546        while let Some(chunk) = body.next().await {
547            let chunk = chunk?;
548            out.write_all(&chunk).await?;
549            done += chunk.len() as u64;
550            progress(done, total);
551        }
552        out.flush().await?;
553        if total.is_some_and(|total| total != done) {
554            return Err(Error::Transport("the download broke off".into()));
555        }
556        Ok(done)
557    }
558
559    /// Uploads content of `size` bytes (RFC 8620, 6.1); `progress` hears the bytes sent so far.
560    pub async fn upload(
561        &self,
562        account: &str,
563        kind: &str,
564        content: impl AsyncRead + Send + 'static,
565        size: u64,
566        mut progress: impl FnMut(u64, u64) + Send + 'static,
567    ) -> Result<Upload> {
568        let url = self.here(&self.session.upload_url.replace("{accountId}", &encode(account)));
569        let mut done = 0;
570        let stream = tokio_util::io::ReaderStream::with_capacity(content, 256 << 10).inspect_ok(move |chunk| {
571            done += chunk.len() as u64;
572            progress(done, size);
573        });
574        let response = self
575            .streaming
576            .post(url)
577            .header("authorization", self.authorization())
578            .header("content-type", kind)
579            .header("content-length", size.to_string())
580            .body(reqwest::Body::wrap_stream(stream))
581            .send()
582            .await?;
583        let body = check(response, 201).await?;
584        serde_json::from_str(&body).map_err(|error| Error::Protocol(format!("the upload: {error}")))
585    }
586
587    /// Opens the push stream (RFC 8620, 7.3) for all types, with a ping each `ping` seconds: its bytes as they come.
588    pub async fn event_source(
589        &self,
590        ping: u64,
591    ) -> Result<impl Stream<Item = std::io::Result<bytes::Bytes>> + Send + Unpin + use<>> {
592        let url = self
593            .session
594            .event_source_url
595            .replace("{types}", "*")
596            .replace("{closeafter}", "no")
597            .replace("{ping}", &ping.to_string());
598        let response = self
599            .streaming
600            .get(self.here(&url))
601            .header("authorization", self.authorization())
602            .header("accept", "text/event-stream")
603            .send()
604            .await?;
605        match response.status().as_u16() {
606            200 => Ok(response.bytes_stream().map_err(std::io::Error::other)),
607            401 => Err(Error::Unauthorized),
608            status => Err(Error::Status(status, None)),
609        }
610    }
611}
612
613/// The body of an answer, refused beyond `limit` bytes.
614async fn limited_text(response: reqwest::Response, limit: usize) -> Result<String> {
615    let mut body = response.bytes_stream();
616    let mut out = Vec::new();
617    while let Some(chunk) = body.next().await {
618        let chunk = chunk?;
619        if out.len() + chunk.len() > limit {
620            return Err(Error::Protocol(format!("an answer of more than {limit} bytes")));
621        }
622        out.extend_from_slice(&chunk);
623    }
624    String::from_utf8(out).map_err(|_| Error::Protocol("an answer that is not UTF-8".into()))
625}
626
627/// The body of an answer with the status expected; else the error it amounts to.
628async fn check(response: reqwest::Response, expected: u16) -> Result<String> {
629    let status = response.status().as_u16();
630    let body = limited_text(response, 16 << 20).await.unwrap_or_default();
631    match status {
632        401 => Err(Error::Unauthorized),
633        status if status == expected || (expected == 201 && status == 200) => Ok(body),
634        status => Err(Error::Status(status, problem_detail(&body))),
635    }
636}
637
638/// The first refused change of a `*/set` response, as an error.
639pub fn set_error(result: &Value) -> Result<()> {
640    for key in ["notCreated", "notUpdated", "notDestroyed"] {
641        if let Some(first) = result.get(key).and_then(Value::as_object).and_then(|failed| failed.values().next()) {
642            return Err(Error::Set {
643                kind: first.get("type").and_then(Value::as_str).unwrap_or("serverFail").to_owned(),
644                description: first.get("description").and_then(Value::as_str).map(str::to_owned),
645            });
646        }
647    }
648    Ok(())
649}
650
651/// The answer to an upload (RFC 8620, 6.1).
652#[derive(Debug, Clone, Deserialize)]
653#[serde(rename_all = "camelCase")]
654pub struct Upload {
655    pub account_id: String,
656    pub blob_id: String,
657    #[serde(rename = "type", default)]
658    pub kind: String,
659    pub size: u64,
660}
661
662/// `#ids` referring to the `/ids` of a query call (RFC 8620, 3.7).
663pub fn ids_of(call: &str, name: &str) -> Value {
664    json!({ "resultOf": call, "name": name, "path": "/ids" })
665}
666
667#[cfg(test)]
668mod tests {
669    use super::*;
670
671    #[test]
672    fn origins() {
673        assert_eq!(split_origin("https://a.example/jmap/api"), Some(("https://a.example", "/jmap/api")));
674        assert_eq!(split_origin("http://[::1]:8080"), Some(("http://[::1]:8080", "/")));
675        assert_eq!(split_origin("https://a.example?x"), Some(("https://a.example", "?x")));
676        assert_eq!(split_origin("/jmap/api"), None);
677        assert_eq!(split_origin("https:///x"), None);
678        assert!(is_local("http://localhost:8080"));
679        assert!(is_local("http://login.example.localhost"));
680        assert!(is_local("http://127.0.0.1:1"));
681        assert!(is_local("http://[::1]:1"));
682        assert!(!is_local("http://example.org"));
683        assert!(!is_local("http://localhost.example.org"));
684    }
685
686    #[test]
687    fn session_urls() {
688        assert_eq!(session_url(None, "a@firma.example").unwrap(), "https://firma.example/.well-known/jmap");
689        assert_eq!(session_url(Some("login.example"), "a").unwrap(), "https://login.example/.well-known/jmap");
690        assert_eq!(session_url(Some("http://localhost:8080/"), "a").unwrap(), "http://localhost:8080/.well-known/jmap");
691        assert_eq!(session_url(Some("https://x/jmap/session"), "a").unwrap(), "https://x/jmap/session");
692        assert!(session_url(None, "nodomain").is_err());
693    }
694
695    #[test]
696    fn encoding() {
697        assert_eq!(encode("a b/ä.txt"), "a%20b%2F%C3%A4.txt");
698        assert_eq!(encode("text/plain"), "text%2Fplain");
699    }
700
701    #[test]
702    fn set_errors() {
703        assert!(set_error(&json!({ "created": { "a": {} } })).is_ok());
704        let error = set_error(&json!({ "notUpdated": { "x": { "type": "forbidden", "description": "no" } } }));
705        assert!(matches!(error, Err(Error::Set { kind, .. }) if kind == "forbidden"));
706    }
707}
kernoid_client/lib.rs

kernoid_client/
lib.rs
