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

kernoid_client/
lib.rs
