1pub mod calendars;
8pub mod contacts;
9pub mod e2e;
10pub mod files;
11pub mod mail;
12pub mod oidc;
13pub mod push;
14pub mod reply;
15pub mod srv;
16pub mod transfer;
17
18use std::{
19 collections::BTreeMap,
20 fmt,
21 net::{IpAddr, Ipv4Addr, SocketAddr},
22 time::Duration,
23};
24
25use base64::Engine;
26use futures_util::{Stream, StreamExt, TryStreamExt};
27use serde::Deserialize;
28use serde_json::{Map, Value, json};
29use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt};
30
31pub const CORE: &str = "urn:ietf:params:jmap:core";
32pub const MAIL: &str = "urn:ietf:params:jmap:mail";
33pub const SUBMISSION: &str = "urn:ietf:params:jmap:submission";
34pub const CALENDARS: &str = "urn:ietf:params:jmap:calendars";
35pub const CONTACTS: &str = "urn:ietf:params:jmap:contacts";
36pub const FILENODE: &str = "urn:ietf:params:jmap:filenode";
37pub const FILE_VERSIONS: &str = "https://kernoid.dev/jmap/file-versions";
39pub const FILE_LINKS: &str = "https://kernoid.dev/jmap/file-links";
41pub const PRINCIPALS: &str = "urn:ietf:params:jmap:principals";
42pub const VACATION: &str = "urn:ietf:params:jmap:vacationresponse";
43
44#[derive(Debug)]
46pub enum Error {
47 Unauthorized,
49 Status(u16, Option<String>),
51 Transport(String),
53 Method { kind: String, description: Option<String> },
55 Set { kind: String, description: Option<String> },
57 Protocol(String),
59 Io(std::io::Error),
61 E2e(String),
63}
64
65impl fmt::Display for Error {
66 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
67 match self {
68 Self::Unauthorized => write!(f, "the address or password was not accepted"),
69 Self::Status(status, Some(detail)) => write!(f, "the server answered {status}: {detail}"),
70 Self::Status(status, None) => write!(f, "the server answered {status}"),
71 Self::Transport(error) => write!(f, "connection: {error}"),
72 Self::Method { kind, description: Some(description) }
73 | Self::Set { kind, description: Some(description) } => {
74 write!(f, "{kind}: {description}")
75 }
76 Self::Method { kind, description: None } | Self::Set { kind, description: None } => write!(f, "{kind}"),
77 Self::Protocol(error) => write!(f, "unexpected answer: {error}"),
78 Self::Io(error) => write!(f, "{error}"),
79 Self::E2e(error) => write!(f, "end-to-end encryption: {error}"),
80 }
81 }
82}
83
84impl std::error::Error for Error {}
85
86impl From<reqwest::Error> for Error {
87 fn from(error: reqwest::Error) -> Self {
88 Self::Transport(error.to_string())
89 }
90}
91
92impl From<kernoid_e2e::Error> for Error {
93 fn from(error: kernoid_e2e::Error) -> Self {
94 Self::E2e(error.to_string())
95 }
96}
97
98impl From<std::io::Error> for Error {
99 fn from(error: std::io::Error) -> Self {
100 Self::Io(error)
101 }
102}
103
104pub type Result<T> = std::result::Result<T, Error>;
105
106#[derive(Debug, Clone, Deserialize)]
108#[serde(rename_all = "camelCase")]
109pub struct Account {
110 pub name: String,
111 #[serde(default)]
112 pub is_personal: bool,
113 #[serde(default)]
114 pub is_read_only: bool,
115 #[serde(default)]
116 pub account_capabilities: Map<String, Value>,
117 #[serde(default, rename = "kernoid:drive")]
119 pub drive: bool,
120}
121
122#[derive(Debug, Clone, Deserialize)]
124#[serde(rename_all = "camelCase")]
125pub struct Session {
126 #[serde(default)]
127 pub capabilities: Map<String, Value>,
128 pub accounts: BTreeMap<String, Account>,
129 #[serde(default)]
130 pub primary_accounts: BTreeMap<String, String>,
131 pub username: String,
132 pub api_url: String,
133 pub download_url: String,
134 pub upload_url: String,
135 #[serde(default)]
136 pub event_source_url: String,
137 #[serde(default)]
138 pub state: String,
139}
140
141impl Session {
142 pub fn primary(&self, capability: &str) -> Option<&str> {
144 self.primary_accounts.get(capability).map(String::as_str)
145 }
146
147 pub fn accounts_with(&self, capability: &str) -> Vec<(&str, &Account)> {
149 let primary = self.primary(capability);
150 let mut out: Vec<_> = self
151 .accounts
152 .iter()
153 .filter(|(_, account)| account.account_capabilities.contains_key(capability))
154 .map(|(id, account)| (id.as_str(), account))
155 .collect();
156 out.sort_by(|a, b| (Some(a.0) != primary, &a.1.name).cmp(&(Some(b.0) != primary, &b.1.name)));
157 out
158 }
159}
160
161#[derive(Debug, Clone)]
163pub struct Responses(pub Vec<(String, Value, String)>);
164
165impl Responses {
166 pub fn get(&self, id: &str) -> Option<&Value> {
168 self.0.iter().find(|(_, _, call)| call == id).map(|(_, value, _)| value)
169 }
170
171 pub fn parse<T: serde::de::DeserializeOwned>(&self, id: &str) -> Result<T> {
173 let value = self.get(id).ok_or_else(|| Error::Protocol(format!("no response to {id}")))?;
174 serde_json::from_value(value.clone()).map_err(|error| Error::Protocol(format!("{id}: {error}")))
175 }
176}
177
178pub fn split_origin(url: &str) -> Option<(&str, &str)> {
180 let (scheme, after) = url.split_once("://")?;
181 if scheme.is_empty() || !scheme.bytes().all(|b| b.is_ascii_alphanumeric() || b"+-.".contains(&b)) {
182 return None;
183 }
184 let end = after.find(['/', '?', '#']).unwrap_or(after.len());
185 if end == 0 {
186 return None;
187 }
188 let origin = &url[..scheme.len() + 3 + end];
189 let rest = &after[end..];
190 Some((origin, if rest.is_empty() { "/" } else { rest }))
191}
192
193fn host(origin: &str) -> &str {
195 let authority = origin.split_once("://").map_or(origin, |(_, rest)| rest);
196 let authority = authority.rsplit_once('@').map_or(authority, |(_, host)| host);
197 if let Some(rest) = authority.strip_prefix('[') {
198 return rest.split_once(']').map_or(rest, |(host, _)| host);
199 }
200 authority.split_once(':').map_or(authority, |(host, _)| host)
201}
202
203pub fn is_local(origin: &str) -> bool {
205 let host = host(origin).to_ascii_lowercase();
206 host == "localhost"
207 || host.ends_with(".localhost")
208 || host.parse::<std::net::IpAddr>().is_ok_and(|address| address.is_loopback())
209}
210
211pub fn encode(value: &str) -> String {
213 let mut out = String::with_capacity(value.len());
214 for byte in value.bytes() {
215 if byte.is_ascii_alphanumeric() || b"-._~".contains(&byte) {
216 out.push(byte as char);
217 } else {
218 out.push_str(&format!("%{byte:02X}"));
219 }
220 }
221 out
222}
223
224pub fn session_url(server: Option<&str>, address: &str) -> Result<String> {
227 let server = match server {
228 Some(server) if server.contains("://") => server.trim_end_matches('/').to_owned(),
229 Some(server) => format!("https://{}", server.trim_end_matches('/')),
230 None => {
231 let domain = address
232 .rsplit_once('@')
233 .map(|(_, domain)| domain)
234 .filter(|domain| !domain.is_empty())
235 .ok_or_else(|| Error::Protocol("give a server, or an address with a domain".into()))?;
236 format!("https://{domain}")
237 }
238 };
239 let (_, path) = split_origin(&server).ok_or_else(|| Error::Protocol(format!("not a URL: {server}")))?;
240 Ok(if path == "/" { format!("{server}/.well-known/jmap") } else { server })
241}
242
243#[derive(Debug, Clone)]
245pub struct Options {
246 pub allow_http: bool,
248 pub user_agent: String,
250}
251
252impl Default for Options {
253 fn default() -> Self {
254 Self { allow_http: false, user_agent: concat!("kernoid-client/", env!("CARGO_PKG_VERSION")).into() }
255 }
256}
257
258#[derive(Clone)]
260pub struct Client {
261 http: reqwest::Client,
262 streaming: reqwest::Client,
264 authorization: std::sync::Arc<std::sync::RwLock<String>>,
266 origin: String,
267 session_url: String,
268 pub session: Session,
269}
270
271impl fmt::Debug for Client {
272 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
273 f.debug_struct("Client").field("origin", &self.origin).field("username", &self.session.username).finish()
274 }
275}
276
277fn problem_detail(body: &str) -> Option<String> {
279 let value: Value = serde_json::from_str(body).ok()?;
280 value.get("detail").or_else(|| value.get("description")).and_then(Value::as_str).map(str::to_owned)
281}
282
283fn http(origin: &str, user_agent: &str, timeout: Option<Duration>) -> Result<reqwest::Client> {
287 let mut builder = reqwest::Client::builder()
288 .redirect(reqwest::redirect::Policy::limited(5))
289 .connect_timeout(Duration::from_secs(15))
290 .user_agent(user_agent);
291 if let Some(timeout) = timeout {
292 builder = builder.timeout(timeout);
293 }
294 let name = host(origin);
296 if name.ends_with(".localhost") {
297 builder = builder.resolve(name, SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0));
298 }
299 Ok(builder.build()?)
300}
301
302impl Client {
303 pub async fn connect(session_url: &str, username: &str, password: &str, options: &Options) -> Result<Self> {
305 let credentials = base64::engine::general_purpose::STANDARD.encode(format!("{username}:{password}"));
306 Self::open(session_url, username, format!("Basic {credentials}"), options).await
307 }
308
309 pub async fn connect_with_token(session_url: &str, access_token: &str, options: &Options) -> Result<Self> {
312 Self::open(session_url, "", format!("Bearer {access_token}"), options).await
313 }
314
315 pub fn set_access_token(&self, access_token: &str) {
317 *self.authorization.write().unwrap_or_else(|poisoned| poisoned.into_inner()) = format!("Bearer {access_token}");
318 }
319
320 fn authorization(&self) -> String {
321 self.authorization.read().unwrap_or_else(|poisoned| poisoned.into_inner()).clone()
322 }
323
324 async fn open(session_url: &str, username: &str, authorization: String, options: &Options) -> Result<Self> {
325 let (origin, _) =
326 split_origin(session_url).ok_or_else(|| Error::Protocol(format!("not a URL: {session_url}")))?;
327 let origin = origin.to_owned();
328 if !origin.starts_with("https://")
329 && !(options.allow_http && origin.starts_with("http://"))
330 && !is_local(&origin)
331 {
332 return Err(Error::Protocol(format!("{origin} is not HTTPS; the password would travel readable")));
333 }
334 let mut client = Self {
335 http: http(&origin, &options.user_agent, Some(Duration::from_secs(120)))?,
336 streaming: http(&origin, &options.user_agent, None)?,
337 authorization: std::sync::Arc::new(std::sync::RwLock::new(authorization)),
338 origin,
339 session_url: session_url.to_owned(),
340 session: Session {
341 capabilities: Map::new(),
342 accounts: BTreeMap::new(),
343 primary_accounts: BTreeMap::new(),
344 username: username.to_owned(),
345 api_url: String::new(),
346 download_url: String::new(),
347 upload_url: String::new(),
348 event_source_url: String::new(),
349 state: String::new(),
350 },
351 };
352 client.session = client.fetch_session(session_url).await?;
353 Ok(client)
354 }
355
356 pub async fn refresh_session(&mut self) -> Result<()> {
358 self.session = self.fetch_session(&self.session_url.clone()).await?;
359 Ok(())
360 }
361
362 async fn fetch_session(&self, url: &str) -> Result<Session> {
363 let response = self
364 .http
365 .get(url)
366 .header("authorization", self.authorization())
367 .header("accept", "application/json")
368 .send()
369 .await?;
370 let body = check(response, 200).await?;
371 serde_json::from_str(&body).map_err(|error| Error::Protocol(format!("the session: {error}")))
372 }
373
374 pub fn origin(&self) -> &str {
376 &self.origin
377 }
378
379 fn here(&self, url: &str) -> String {
382 match split_origin(url) {
383 Some((_, rest)) => format!("{}{rest}", self.origin),
384 None if url.starts_with('/') => format!("{}{url}", self.origin),
385 None => format!("{}/{url}", self.origin),
386 }
387 }
388
389 pub fn primary(&self, capability: &str) -> Result<String> {
391 self.session
392 .primary(capability)
393 .map(str::to_owned)
394 .ok_or_else(|| Error::Method { kind: "accountNotFound".into(), description: Some(capability.into()) })
395 }
396
397 pub async fn request(&self, using: &[&str], calls: Vec<(&str, Value, &str)>) -> Result<Responses> {
399 let mut capabilities = vec![CORE];
400 capabilities.extend(using.iter().filter(|capability| **capability != CORE));
401 let calls: Vec<Value> = calls.into_iter().map(|(name, arguments, id)| json!([name, arguments, id])).collect();
402 let body = json!({ "using": capabilities, "methodCalls": calls }).to_string();
403 let response = self
404 .http
405 .post(self.here(&self.session.api_url))
406 .header("authorization", self.authorization())
407 .header("content-type", "application/json")
408 .header("accept", "application/json")
409 .body(body)
410 .send()
411 .await?;
412 let status = response.status().as_u16();
413 let text = limited_text(response, 64 << 20).await?;
414 if status == 401 {
415 return Err(Error::Unauthorized);
416 }
417 let value: Value = serde_json::from_str(&text).map_err(|_| Error::Status(status, problem_detail(&text)))?;
418 if status != 200 {
419 let kind = value.get("type").and_then(Value::as_str).unwrap_or("serverFail");
421 let kind = kind.rsplit(':').next().unwrap_or(kind).to_owned();
422 return Err(Error::Method { kind, description: problem_detail(&text) });
423 }
424 let responses = value
425 .get("methodResponses")
426 .and_then(Value::as_array)
427 .ok_or_else(|| Error::Protocol("no methodResponses".into()))?;
428 let mut out = Vec::with_capacity(responses.len());
429 for response in responses {
430 let (Some(name), Some(arguments), Some(id)) =
431 (response.get(0).and_then(Value::as_str), response.get(1), response.get(2).and_then(Value::as_str))
432 else {
433 return Err(Error::Protocol("a malformed method response".into()));
434 };
435 if name == "error" {
436 return Err(Error::Method {
437 kind: arguments.get("type").and_then(Value::as_str).unwrap_or("serverFail").to_owned(),
438 description: arguments.get("description").and_then(Value::as_str).map(str::to_owned),
439 });
440 }
441 out.push((name.to_owned(), arguments.clone(), id.to_owned()));
442 }
443 Ok(Responses(out))
444 }
445
446 pub async fn call(&self, using: &[&str], name: &str, arguments: Value) -> Result<Value> {
448 let responses = self.request(using, vec![(name, arguments, "0")]).await?;
449 responses.get("0").cloned().ok_or_else(|| Error::Protocol(format!("no response to {name}")))
450 }
451
452 pub async fn set(&self, using: &[&str], kind: &str, account: &str, change: Value, extra: Value) -> Result<Value> {
455 let mut arguments = json!({ "accountId": account });
456 for source in [change, extra] {
457 if let (Some(target), Value::Object(source)) = (arguments.as_object_mut(), source) {
458 target.extend(source);
459 }
460 }
461 let result = self.call(using, &format!("{kind}/set"), arguments).await?;
462 set_error(&result)?;
463 Ok(result)
464 }
465
466 pub fn download_url(&self, account: &str, blob: &str, name: &str, kind: &str) -> String {
468 let url = self
469 .session
470 .download_url
471 .replace("{accountId}", &encode(account))
472 .replace("{blobId}", &encode(blob))
473 .replace("{name}", &encode(name))
474 .replace("{type}", &encode(kind));
475 self.here(&url)
476 }
477
478 pub async fn download(
480 &self,
481 account: &str,
482 blob: &str,
483 name: &str,
484 kind: &str,
485 out: &mut (dyn AsyncWrite + Unpin + Send),
486 progress: &mut (dyn FnMut(u64, Option<u64>) + Send),
487 ) -> Result<u64> {
488 let url = self.download_url(account, blob, name, kind);
489 let response = self.streaming.get(url).header("authorization", self.authorization()).send().await?;
490 let status = response.status().as_u16();
491 if status == 401 {
492 return Err(Error::Unauthorized);
493 }
494 if status != 200 {
495 let body = limited_text(response, 16 << 20).await.unwrap_or_default();
496 return Err(Error::Status(status, problem_detail(&body)));
497 }
498 let total = response.content_length();
499 let mut body = response.bytes_stream();
500 let mut done = 0;
501 while let Some(chunk) = body.next().await {
502 let chunk = chunk?;
503 out.write_all(&chunk).await?;
504 done += chunk.len() as u64;
505 progress(done, total);
506 }
507 out.flush().await?;
508 if total.is_some_and(|total| total != done) {
509 return Err(Error::Transport("the download broke off".into()));
510 }
511 Ok(done)
512 }
513
514 pub async fn upload(
516 &self,
517 account: &str,
518 kind: &str,
519 content: impl AsyncRead + Send + 'static,
520 size: u64,
521 mut progress: impl FnMut(u64, u64) + Send + 'static,
522 ) -> Result<Upload> {
523 let url = self.here(&self.session.upload_url.replace("{accountId}", &encode(account)));
524 let mut done = 0;
525 let stream = tokio_util::io::ReaderStream::with_capacity(content, 256 << 10).inspect_ok(move |chunk| {
526 done += chunk.len() as u64;
527 progress(done, size);
528 });
529 let response = self
530 .streaming
531 .post(url)
532 .header("authorization", self.authorization())
533 .header("content-type", kind)
534 .header("content-length", size.to_string())
535 .body(reqwest::Body::wrap_stream(stream))
536 .send()
537 .await?;
538 let body = check(response, 201).await?;
539 serde_json::from_str(&body).map_err(|error| Error::Protocol(format!("the upload: {error}")))
540 }
541
542 pub async fn event_source(
544 &self,
545 ping: u64,
546 ) -> Result<impl Stream<Item = std::io::Result<bytes::Bytes>> + Send + Unpin + use<>> {
547 let url = self
548 .session
549 .event_source_url
550 .replace("{types}", "*")
551 .replace("{closeafter}", "no")
552 .replace("{ping}", &ping.to_string());
553 let response = self
554 .streaming
555 .get(self.here(&url))
556 .header("authorization", self.authorization())
557 .header("accept", "text/event-stream")
558 .send()
559 .await?;
560 match response.status().as_u16() {
561 200 => Ok(response.bytes_stream().map_err(std::io::Error::other)),
562 401 => Err(Error::Unauthorized),
563 status => Err(Error::Status(status, None)),
564 }
565 }
566}
567
568async fn limited_text(response: reqwest::Response, limit: usize) -> Result<String> {
570 let mut body = response.bytes_stream();
571 let mut out = Vec::new();
572 while let Some(chunk) = body.next().await {
573 let chunk = chunk?;
574 if out.len() + chunk.len() > limit {
575 return Err(Error::Protocol(format!("an answer of more than {limit} bytes")));
576 }
577 out.extend_from_slice(&chunk);
578 }
579 String::from_utf8(out).map_err(|_| Error::Protocol("an answer that is not UTF-8".into()))
580}
581
582async fn check(response: reqwest::Response, expected: u16) -> Result<String> {
584 let status = response.status().as_u16();
585 let body = limited_text(response, 16 << 20).await.unwrap_or_default();
586 match status {
587 401 => Err(Error::Unauthorized),
588 status if status == expected || (expected == 201 && status == 200) => Ok(body),
589 status => Err(Error::Status(status, problem_detail(&body))),
590 }
591}
592
593pub fn set_error(result: &Value) -> Result<()> {
595 for key in ["notCreated", "notUpdated", "notDestroyed"] {
596 if let Some(first) = result.get(key).and_then(Value::as_object).and_then(|failed| failed.values().next()) {
597 return Err(Error::Set {
598 kind: first.get("type").and_then(Value::as_str).unwrap_or("serverFail").to_owned(),
599 description: first.get("description").and_then(Value::as_str).map(str::to_owned),
600 });
601 }
602 }
603 Ok(())
604}
605
606#[derive(Debug, Clone, Deserialize)]
608#[serde(rename_all = "camelCase")]
609pub struct Upload {
610 pub account_id: String,
611 pub blob_id: String,
612 #[serde(rename = "type", default)]
613 pub kind: String,
614 pub size: u64,
615}
616
617pub fn ids_of(call: &str, name: &str) -> Value {
619 json!({ "resultOf": call, "name": name, "path": "/ids" })
620}
621
622#[cfg(test)]
623mod tests {
624 use super::*;
625
626 #[test]
627 fn origins() {
628 assert_eq!(split_origin("https://a.example/jmap/api"), Some(("https://a.example", "/jmap/api")));
629 assert_eq!(split_origin("http://[::1]:8080"), Some(("http://[::1]:8080", "/")));
630 assert_eq!(split_origin("https://a.example?x"), Some(("https://a.example", "?x")));
631 assert_eq!(split_origin("/jmap/api"), None);
632 assert_eq!(split_origin("https:///x"), None);
633 assert!(is_local("http://localhost:8080"));
634 assert!(is_local("http://login.example.localhost"));
635 assert!(is_local("http://127.0.0.1:1"));
636 assert!(is_local("http://[::1]:1"));
637 assert!(!is_local("http://example.org"));
638 assert!(!is_local("http://localhost.example.org"));
639 }
640
641 #[test]
642 fn session_urls() {
643 assert_eq!(session_url(None, "a@firma.example").unwrap(), "https://firma.example/.well-known/jmap");
644 assert_eq!(session_url(Some("login.example"), "a").unwrap(), "https://login.example/.well-known/jmap");
645 assert_eq!(session_url(Some("http://localhost:8080/"), "a").unwrap(), "http://localhost:8080/.well-known/jmap");
646 assert_eq!(session_url(Some("https://x/jmap/session"), "a").unwrap(), "https://x/jmap/session");
647 assert!(session_url(None, "nodomain").is_err());
648 }
649
650 #[test]
651 fn encoding() {
652 assert_eq!(encode("a b/ä.txt"), "a%20b%2F%C3%A4.txt");
653 assert_eq!(encode("text/plain"), "text%2Fplain");
654 }
655
656 #[test]
657 fn set_errors() {
658 assert!(set_error(&json!({ "created": { "a": {} } })).is_ok());
659 let error = set_error(&json!({ "notUpdated": { "x": { "type": "forbidden", "description": "no" } } }));
660 assert!(matches!(error, Err(Error::Set { kind, .. }) if kind == "forbidden"));
661 }
662}