pub mod calendars;
pub mod contacts;
pub mod e2e;
pub mod files;
pub mod mail;
pub mod oidc;
pub mod push;
pub mod reply;
pub mod srv;
pub mod transfer;
use std::{
collections::BTreeMap,
fmt,
net::{IpAddr, Ipv4Addr, SocketAddr},
time::Duration,
};
use base64::Engine;
use futures_util::{Stream, StreamExt, TryStreamExt};
use serde::Deserialize;
use serde_json::{Map, Value, json};
use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt};
pub const CORE: &str = "urn:ietf:params:jmap:core";
pub const MAIL: &str = "urn:ietf:params:jmap:mail";
pub const SUBMISSION: &str = "urn:ietf:params:jmap:submission";
pub const CALENDARS: &str = "urn:ietf:params:jmap:calendars";
pub const CONTACTS: &str = "urn:ietf:params:jmap:contacts";
pub const FILENODE: &str = "urn:ietf:params:jmap:filenode";
pub const FILE_VERSIONS: &str = "https://kernoid.dev/jmap/file-versions";
pub const FILE_LINKS: &str = "https://kernoid.dev/jmap/file-links";
pub const PRINCIPALS: &str = "urn:ietf:params:jmap:principals";
pub const VACATION: &str = "urn:ietf:params:jmap:vacationresponse";
#[derive(Debug)]
pub enum Error {
Unauthorized,
Status(u16, Option<String>),
Transport(String),
Method { kind: String, description: Option<String> },
Set { kind: String, description: Option<String> },
Protocol(String),
Io(std::io::Error),
E2e(String),
}
impl fmt::Display for Error {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Unauthorized => write!(f, "the address or password was not accepted"),
Self::Status(status, Some(detail)) => write!(f, "the server answered {status}: {detail}"),
Self::Status(status, None) => write!(f, "the server answered {status}"),
Self::Transport(error) => write!(f, "connection: {error}"),
Self::Method { kind, description: Some(description) }
| Self::Set { kind, description: Some(description) } => {
write!(f, "{kind}: {description}")
}
Self::Method { kind, description: None } | Self::Set { kind, description: None } => write!(f, "{kind}"),
Self::Protocol(error) => write!(f, "unexpected answer: {error}"),
Self::Io(error) => write!(f, "{error}"),
Self::E2e(error) => write!(f, "end-to-end encryption: {error}"),
}
}
}
impl std::error::Error for Error {}
impl From<reqwest::Error> for Error {
fn from(error: reqwest::Error) -> Self {
Self::Transport(error.to_string())
}
}
impl From<kernoid_e2e::Error> for Error {
fn from(error: kernoid_e2e::Error) -> Self {
Self::E2e(error.to_string())
}
}
impl From<std::io::Error> for Error {
fn from(error: std::io::Error) -> Self {
Self::Io(error)
}
}
pub type Result<T> = std::result::Result<T, Error>;
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct Account {
pub name: String,
#[serde(default)]
pub is_personal: bool,
#[serde(default)]
pub is_read_only: bool,
#[serde(default)]
pub account_capabilities: Map<String, Value>,
#[serde(default, rename = "kernoid:drive")]
pub drive: bool,
}
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct Session {
#[serde(default)]
pub capabilities: Map<String, Value>,
pub accounts: BTreeMap<String, Account>,
#[serde(default)]
pub primary_accounts: BTreeMap<String, String>,
pub username: String,
pub api_url: String,
pub download_url: String,
pub upload_url: String,
#[serde(default)]
pub event_source_url: String,
#[serde(default)]
pub state: String,
}
impl Session {
pub fn primary(&self, capability: &str) -> Option<&str> {
self.primary_accounts.get(capability).map(String::as_str)
}
pub fn accounts_with(&self, capability: &str) -> Vec<(&str, &Account)> {
let primary = self.primary(capability);
let mut out: Vec<_> = self
.accounts
.iter()
.filter(|(_, account)| account.account_capabilities.contains_key(capability))
.map(|(id, account)| (id.as_str(), account))
.collect();
out.sort_by(|a, b| (Some(a.0) != primary, &a.1.name).cmp(&(Some(b.0) != primary, &b.1.name)));
out
}
}
#[derive(Debug, Clone)]
pub struct Responses(pub Vec<(String, Value, String)>);
impl Responses {
pub fn get(&self, id: &str) -> Option<&Value> {
self.0.iter().find(|(_, _, call)| call == id).map(|(_, value, _)| value)
}
pub fn parse<T: serde::de::DeserializeOwned>(&self, id: &str) -> Result<T> {
let value = self.get(id).ok_or_else(|| Error::Protocol(format!("no response to {id}")))?;
serde_json::from_value(value.clone()).map_err(|error| Error::Protocol(format!("{id}: {error}")))
}
}
pub fn split_origin(url: &str) -> Option<(&str, &str)> {
let (scheme, after) = url.split_once("://")?;
if scheme.is_empty() || !scheme.bytes().all(|b| b.is_ascii_alphanumeric() || b"+-.".contains(&b)) {
return None;
}
let end = after.find(['/', '?', '#']).unwrap_or(after.len());
if end == 0 {
return None;
}
let origin = &url[..scheme.len() + 3 + end];
let rest = &after[end..];
Some((origin, if rest.is_empty() { "/" } else { rest }))
}
fn host(origin: &str) -> &str {
let authority = origin.split_once("://").map_or(origin, |(_, rest)| rest);
let authority = authority.rsplit_once('@').map_or(authority, |(_, host)| host);
if let Some(rest) = authority.strip_prefix('[') {
return rest.split_once(']').map_or(rest, |(host, _)| host);
}
authority.split_once(':').map_or(authority, |(host, _)| host)
}
pub fn is_local(origin: &str) -> bool {
let host = host(origin).to_ascii_lowercase();
host == "localhost"
|| host.ends_with(".localhost")
|| host.parse::<std::net::IpAddr>().is_ok_and(|address| address.is_loopback())
}
pub fn encode(value: &str) -> String {
let mut out = String::with_capacity(value.len());
for byte in value.bytes() {
if byte.is_ascii_alphanumeric() || b"-._~".contains(&byte) {
out.push(byte as char);
} else {
out.push_str(&format!("%{byte:02X}"));
}
}
out
}
pub fn session_url(server: Option<&str>, address: &str) -> Result<String> {
let server = match server {
Some(server) if server.contains("://") => server.trim_end_matches('/').to_owned(),
Some(server) => format!("https://{}", server.trim_end_matches('/')),
None => {
let domain = address
.rsplit_once('@')
.map(|(_, domain)| domain)
.filter(|domain| !domain.is_empty())
.ok_or_else(|| Error::Protocol("give a server, or an address with a domain".into()))?;
format!("https://{domain}")
}
};
let (_, path) = split_origin(&server).ok_or_else(|| Error::Protocol(format!("not a URL: {server}")))?;
Ok(if path == "/" { format!("{server}/.well-known/jmap") } else { server })
}
#[derive(Debug, Clone)]
pub struct Options {
pub allow_http: bool,
pub user_agent: String,
}
impl Default for Options {
fn default() -> Self {
Self { allow_http: false, user_agent: concat!("kernoid-client/", env!("CARGO_PKG_VERSION")).into() }
}
}
#[derive(Clone)]
pub struct Client {
http: reqwest::Client,
streaming: reqwest::Client,
authorization: std::sync::Arc<std::sync::RwLock<String>>,
origin: String,
session_url: String,
pub session: Session,
}
impl fmt::Debug for Client {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Client").field("origin", &self.origin).field("username", &self.session.username).finish()
}
}
fn problem_detail(body: &str) -> Option<String> {
let value: Value = serde_json::from_str(body).ok()?;
value.get("detail").or_else(|| value.get("description")).and_then(Value::as_str).map(str::to_owned)
}
fn http(origin: &str, user_agent: &str, timeout: Option<Duration>) -> Result<reqwest::Client> {
let mut builder = reqwest::Client::builder()
.redirect(reqwest::redirect::Policy::limited(5))
.connect_timeout(Duration::from_secs(15))
.user_agent(user_agent);
if let Some(timeout) = timeout {
builder = builder.timeout(timeout);
}
let name = host(origin);
if name.ends_with(".localhost") {
builder = builder.resolve(name, SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0));
}
Ok(builder.build()?)
}
impl Client {
pub async fn connect(session_url: &str, username: &str, password: &str, options: &Options) -> Result<Self> {
let credentials = base64::engine::general_purpose::STANDARD.encode(format!("{username}:{password}"));
Self::open(session_url, username, format!("Basic {credentials}"), options).await
}
pub async fn connect_with_token(session_url: &str, access_token: &str, options: &Options) -> Result<Self> {
Self::open(session_url, "", format!("Bearer {access_token}"), options).await
}
pub fn set_access_token(&self, access_token: &str) {
*self.authorization.write().unwrap_or_else(|poisoned| poisoned.into_inner()) = format!("Bearer {access_token}");
}
fn authorization(&self) -> String {
self.authorization.read().unwrap_or_else(|poisoned| poisoned.into_inner()).clone()
}
async fn open(session_url: &str, username: &str, authorization: String, options: &Options) -> Result<Self> {
let (origin, _) =
split_origin(session_url).ok_or_else(|| Error::Protocol(format!("not a URL: {session_url}")))?;
let origin = origin.to_owned();
if !origin.starts_with("https://")
&& !(options.allow_http && origin.starts_with("http://"))
&& !is_local(&origin)
{
return Err(Error::Protocol(format!("{origin} is not HTTPS; the password would travel readable")));
}
let mut client = Self {
http: http(&origin, &options.user_agent, Some(Duration::from_secs(120)))?,
streaming: http(&origin, &options.user_agent, None)?,
authorization: std::sync::Arc::new(std::sync::RwLock::new(authorization)),
origin,
session_url: session_url.to_owned(),
session: Session {
capabilities: Map::new(),
accounts: BTreeMap::new(),
primary_accounts: BTreeMap::new(),
username: username.to_owned(),
api_url: String::new(),
download_url: String::new(),
upload_url: String::new(),
event_source_url: String::new(),
state: String::new(),
},
};
client.session = client.fetch_session(session_url).await?;
Ok(client)
}
pub async fn refresh_session(&mut self) -> Result<()> {
self.session = self.fetch_session(&self.session_url.clone()).await?;
Ok(())
}
async fn fetch_session(&self, url: &str) -> Result<Session> {
let response = self
.http
.get(url)
.header("authorization", self.authorization())
.header("accept", "application/json")
.send()
.await?;
let body = check(response, 200).await?;
serde_json::from_str(&body).map_err(|error| Error::Protocol(format!("the session: {error}")))
}
pub fn origin(&self) -> &str {
&self.origin
}
fn here(&self, url: &str) -> String {
match split_origin(url) {
Some((_, rest)) => format!("{}{rest}", self.origin),
None if url.starts_with('/') => format!("{}{url}", self.origin),
None => format!("{}/{url}", self.origin),
}
}
pub fn primary(&self, capability: &str) -> Result<String> {
self.session
.primary(capability)
.map(str::to_owned)
.ok_or_else(|| Error::Method { kind: "accountNotFound".into(), description: Some(capability.into()) })
}
pub async fn request(&self, using: &[&str], calls: Vec<(&str, Value, &str)>) -> Result<Responses> {
let mut capabilities = vec![CORE];
capabilities.extend(using.iter().filter(|capability| **capability != CORE));
let calls: Vec<Value> = calls.into_iter().map(|(name, arguments, id)| json!([name, arguments, id])).collect();
let body = json!({ "using": capabilities, "methodCalls": calls }).to_string();
let response = self
.http
.post(self.here(&self.session.api_url))
.header("authorization", self.authorization())
.header("content-type", "application/json")
.header("accept", "application/json")
.body(body)
.send()
.await?;
let status = response.status().as_u16();
let text = limited_text(response, 64 << 20).await?;
if status == 401 {
return Err(Error::Unauthorized);
}
let value: Value = serde_json::from_str(&text).map_err(|_| Error::Status(status, problem_detail(&text)))?;
if status != 200 {
let kind = value.get("type").and_then(Value::as_str).unwrap_or("serverFail");
let kind = kind.rsplit(':').next().unwrap_or(kind).to_owned();
return Err(Error::Method { kind, description: problem_detail(&text) });
}
let responses = value
.get("methodResponses")
.and_then(Value::as_array)
.ok_or_else(|| Error::Protocol("no methodResponses".into()))?;
let mut out = Vec::with_capacity(responses.len());
for response in responses {
let (Some(name), Some(arguments), Some(id)) =
(response.get(0).and_then(Value::as_str), response.get(1), response.get(2).and_then(Value::as_str))
else {
return Err(Error::Protocol("a malformed method response".into()));
};
if name == "error" {
return Err(Error::Method {
kind: arguments.get("type").and_then(Value::as_str).unwrap_or("serverFail").to_owned(),
description: arguments.get("description").and_then(Value::as_str).map(str::to_owned),
});
}
out.push((name.to_owned(), arguments.clone(), id.to_owned()));
}
Ok(Responses(out))
}
pub async fn call(&self, using: &[&str], name: &str, arguments: Value) -> Result<Value> {
let responses = self.request(using, vec![(name, arguments, "0")]).await?;
responses.get("0").cloned().ok_or_else(|| Error::Protocol(format!("no response to {name}")))
}
pub async fn set(&self, using: &[&str], kind: &str, account: &str, change: Value, extra: Value) -> Result<Value> {
let mut arguments = json!({ "accountId": account });
for source in [change, extra] {
if let (Some(target), Value::Object(source)) = (arguments.as_object_mut(), source) {
target.extend(source);
}
}
let result = self.call(using, &format!("{kind}/set"), arguments).await?;
set_error(&result)?;
Ok(result)
}
pub fn download_url(&self, account: &str, blob: &str, name: &str, kind: &str) -> String {
let url = self
.session
.download_url
.replace("{accountId}", &encode(account))
.replace("{blobId}", &encode(blob))
.replace("{name}", &encode(name))
.replace("{type}", &encode(kind));
self.here(&url)
}
pub async fn download(
&self,
account: &str,
blob: &str,
name: &str,
kind: &str,
out: &mut (dyn AsyncWrite + Unpin + Send),
progress: &mut (dyn FnMut(u64, Option<u64>) + Send),
) -> Result<u64> {
let url = self.download_url(account, blob, name, kind);
let response = self.streaming.get(url).header("authorization", self.authorization()).send().await?;
let status = response.status().as_u16();
if status == 401 {
return Err(Error::Unauthorized);
}
if status != 200 {
let body = limited_text(response, 16 << 20).await.unwrap_or_default();
return Err(Error::Status(status, problem_detail(&body)));
}
let total = response.content_length();
let mut body = response.bytes_stream();
let mut done = 0;
while let Some(chunk) = body.next().await {
let chunk = chunk?;
out.write_all(&chunk).await?;
done += chunk.len() as u64;
progress(done, total);
}
out.flush().await?;
if total.is_some_and(|total| total != done) {
return Err(Error::Transport("the download broke off".into()));
}
Ok(done)
}
pub async fn upload(
&self,
account: &str,
kind: &str,
content: impl AsyncRead + Send + 'static,
size: u64,
mut progress: impl FnMut(u64, u64) + Send + 'static,
) -> Result<Upload> {
let url = self.here(&self.session.upload_url.replace("{accountId}", &encode(account)));
let mut done = 0;
let stream = tokio_util::io::ReaderStream::with_capacity(content, 256 << 10).inspect_ok(move |chunk| {
done += chunk.len() as u64;
progress(done, size);
});
let response = self
.streaming
.post(url)
.header("authorization", self.authorization())
.header("content-type", kind)
.header("content-length", size.to_string())
.body(reqwest::Body::wrap_stream(stream))
.send()
.await?;
let body = check(response, 201).await?;
serde_json::from_str(&body).map_err(|error| Error::Protocol(format!("the upload: {error}")))
}
pub async fn event_source(
&self,
ping: u64,
) -> Result<impl Stream<Item = std::io::Result<bytes::Bytes>> + Send + Unpin + use<>> {
let url = self
.session
.event_source_url
.replace("{types}", "*")
.replace("{closeafter}", "no")
.replace("{ping}", &ping.to_string());
let response = self
.streaming
.get(self.here(&url))
.header("authorization", self.authorization())
.header("accept", "text/event-stream")
.send()
.await?;
match response.status().as_u16() {
200 => Ok(response.bytes_stream().map_err(std::io::Error::other)),
401 => Err(Error::Unauthorized),
status => Err(Error::Status(status, None)),
}
}
}
async fn limited_text(response: reqwest::Response, limit: usize) -> Result<String> {
let mut body = response.bytes_stream();
let mut out = Vec::new();
while let Some(chunk) = body.next().await {
let chunk = chunk?;
if out.len() + chunk.len() > limit {
return Err(Error::Protocol(format!("an answer of more than {limit} bytes")));
}
out.extend_from_slice(&chunk);
}
String::from_utf8(out).map_err(|_| Error::Protocol("an answer that is not UTF-8".into()))
}
async fn check(response: reqwest::Response, expected: u16) -> Result<String> {
let status = response.status().as_u16();
let body = limited_text(response, 16 << 20).await.unwrap_or_default();
match status {
401 => Err(Error::Unauthorized),
status if status == expected || (expected == 201 && status == 200) => Ok(body),
status => Err(Error::Status(status, problem_detail(&body))),
}
}
pub fn set_error(result: &Value) -> Result<()> {
for key in ["notCreated", "notUpdated", "notDestroyed"] {
if let Some(first) = result.get(key).and_then(Value::as_object).and_then(|failed| failed.values().next()) {
return Err(Error::Set {
kind: first.get("type").and_then(Value::as_str).unwrap_or("serverFail").to_owned(),
description: first.get("description").and_then(Value::as_str).map(str::to_owned),
});
}
}
Ok(())
}
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct Upload {
pub account_id: String,
pub blob_id: String,
#[serde(rename = "type", default)]
pub kind: String,
pub size: u64,
}
pub fn ids_of(call: &str, name: &str) -> Value {
json!({ "resultOf": call, "name": name, "path": "/ids" })
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn origins() {
assert_eq!(split_origin("https://a.example/jmap/api"), Some(("https://a.example", "/jmap/api")));
assert_eq!(split_origin("http://[::1]:8080"), Some(("http://[::1]:8080", "/")));
assert_eq!(split_origin("https://a.example?x"), Some(("https://a.example", "?x")));
assert_eq!(split_origin("/jmap/api"), None);
assert_eq!(split_origin("https:///x"), None);
assert!(is_local("http://localhost:8080"));
assert!(is_local("http://login.example.localhost"));
assert!(is_local("http://127.0.0.1:1"));
assert!(is_local("http://[::1]:1"));
assert!(!is_local("http://example.org"));
assert!(!is_local("http://localhost.example.org"));
}
#[test]
fn session_urls() {
assert_eq!(session_url(None, "a@firma.example").unwrap(), "https://firma.example/.well-known/jmap");
assert_eq!(session_url(Some("login.example"), "a").unwrap(), "https://login.example/.well-known/jmap");
assert_eq!(session_url(Some("http://localhost:8080/"), "a").unwrap(), "http://localhost:8080/.well-known/jmap");
assert_eq!(session_url(Some("https://x/jmap/session"), "a").unwrap(), "https://x/jmap/session");
assert!(session_url(None, "nodomain").is_err());
}
#[test]
fn encoding() {
assert_eq!(encode("a b/รค.txt"), "a%20b%2F%C3%A4.txt");
assert_eq!(encode("text/plain"), "text%2Fplain");
}
#[test]
fn set_errors() {
assert!(set_error(&json!({ "created": { "a": {} } })).is_ok());
let error = set_error(&json!({ "notUpdated": { "x": { "type": "forbidden", "description": "no" } } }));
assert!(matches!(error, Err(Error::Set { kind, .. }) if kind == "forbidden"));
}
}