mod integration;
mod maps_integration;
use std::cmp::max;
use std::collections::HashMap;
use std::path::Path;
use anyhow::{anyhow, bail, ensure, format_err, Context as _, Result};
use async_zip::tokio::read::seek::ZipFileReader as SeekZipFileReader;
use deltachat_contact_tools::sanitize_bidi_characters;
use deltachat_derive::FromSql;
use mail_builder::mime::MimePart;
use rusqlite::OptionalExtension;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use sha2::{Digest, Sha256};
use tokio::{fs::File, io::BufReader};
use crate::chat::{self, Chat};
use crate::constants::Chattype;
use crate::contact::ContactId;
use crate::context::Context;
use crate::events::EventType;
use crate::key::{load_self_public_key, DcKey};
use crate::message::{Message, MessageState, MsgId, Viewtype};
use crate::mimefactory::RECOMMENDED_FILE_SIZE;
use crate::mimeparser::SystemMessage;
use crate::param::Param;
use crate::param::Params;
use crate::tools::create_id;
use crate::tools::{create_smeared_timestamp, get_abs_path};
const WEBXDC_API_VERSION: u32 = 1;
pub const WEBXDC_SUFFIX: &str = "xdc";
const WEBXDC_DEFAULT_ICON: &str = "__webxdc__/default-icon.png";
const BODY_DESCR: &str = "Webxdc Status Update";
#[derive(Debug, Deserialize, Default)]
#[non_exhaustive]
pub struct WebxdcManifest {
pub name: Option<String>,
pub min_api: Option<u32>,
pub source_code_url: Option<String>,
pub request_integration: Option<String>,
}
#[derive(Debug, Serialize)]
pub struct WebxdcInfo {
pub name: String,
pub icon: String,
pub document: String,
pub summary: String,
pub source_code_url: String,
pub request_integration: String,
pub internet_access: bool,
pub self_addr: String,
pub send_update_interval: usize,
pub send_update_max_size: usize,
}
#[derive(
Debug,
Copy,
Clone,
Default,
PartialEq,
Eq,
Hash,
PartialOrd,
Ord,
Serialize,
Deserialize,
FromSql,
FromPrimitive,
)]
pub struct StatusUpdateSerial(u32);
impl StatusUpdateSerial {
pub fn new(id: u32) -> StatusUpdateSerial {
StatusUpdateSerial(id)
}
pub const MIN: Self = Self(1);
pub const MAX: Self = Self(u32::MAX - 1);
pub fn to_u32(self) -> u32 {
self.0
}
}
impl rusqlite::types::ToSql for StatusUpdateSerial {
fn to_sql(&self) -> rusqlite::Result<rusqlite::types::ToSqlOutput> {
let val = rusqlite::types::Value::Integer(i64::from(self.0));
let out = rusqlite::types::ToSqlOutput::Owned(val);
Ok(out)
}
}
#[derive(Debug, Deserialize)]
struct StatusUpdates {
updates: Vec<StatusUpdateItem>,
}
#[derive(Debug, Serialize, Deserialize, Default)]
pub struct StatusUpdateItem {
pub payload: Value,
#[serde(skip_serializing_if = "Option::is_none")]
pub info: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub href: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub document: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub summary: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub uid: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub notify: Option<HashMap<String, String>>,
}
#[derive(Debug, Serialize, Deserialize)]
pub(crate) struct StatusUpdateItemAndSerial {
#[serde(flatten)]
item: StatusUpdateItem,
serial: StatusUpdateSerial,
max_serial: StatusUpdateSerial,
}
fn find_zip_entry<'a>(
file: &'a async_zip::ZipFile,
name: &str,
) -> Option<(usize, &'a async_zip::StoredZipEntry)> {
for (i, ent) in file.entries().iter().enumerate() {
if ent.filename().as_bytes() == name.as_bytes() {
return Some((i, ent));
}
}
None
}
const STATUS_UPDATE_SIZE_MAX: usize = 100 << 10;
impl Context {
pub(crate) async fn is_webxdc_file(&self, filename: &str, file: &[u8]) -> Result<bool> {
if !filename.ends_with(WEBXDC_SUFFIX) {
return Ok(false);
}
let archive = match async_zip::base::read::mem::ZipFileReader::new(file.to_vec()).await {
Ok(archive) => archive,
Err(_) => {
info!(self, "{} cannot be opened as zip-file", &filename);
return Ok(false);
}
};
if find_zip_entry(archive.file(), "index.html").is_none() {
info!(self, "{} misses index.html", &filename);
return Ok(false);
}
Ok(true)
}
pub(crate) async fn ensure_sendable_webxdc_file(&self, path: &Path) -> Result<()> {
let filename = path.to_str().unwrap_or_default();
let file = BufReader::new(File::open(path).await?);
let valid = match SeekZipFileReader::with_tokio(file).await {
Ok(archive) => {
if find_zip_entry(archive.file(), "index.html").is_none() {
warn!(self, "{} misses index.html", filename);
false
} else {
true
}
}
Err(_) => {
warn!(self, "{} cannot be opened as zip-file", filename);
false
}
};
if !valid {
bail!("{} is not a valid webxdc file", filename);
}
Ok(())
}
async fn get_overwritable_info_msg_id(
&self,
instance: &Message,
from_id: ContactId,
) -> Result<Option<MsgId>> {
if let Some((last_msg_id, last_from_id, last_param, last_in_repl_to)) = self
.sql
.query_row_optional(
r#"SELECT id, from_id, param, mime_in_reply_to
FROM msgs
WHERE chat_id=?1 AND hidden=0
ORDER BY timestamp DESC, id DESC LIMIT 1"#,
(instance.chat_id,),
|row| {
let last_msg_id: MsgId = row.get(0)?;
let last_from_id: ContactId = row.get(1)?;
let last_param: Params = row.get::<_, String>(2)?.parse().unwrap_or_default();
let last_in_repl_to: String = row.get(3)?;
Ok((last_msg_id, last_from_id, last_param, last_in_repl_to))
},
)
.await?
{
if last_from_id == from_id
&& last_param.get_cmd() == SystemMessage::WebxdcInfoMessage
&& last_in_repl_to == instance.rfc724_mid
{
return Ok(Some(last_msg_id));
}
}
Ok(None)
}
async fn create_status_update_record(
&self,
instance: &Message,
status_update_item: StatusUpdateItem,
timestamp: i64,
can_info_msg: bool,
from_id: ContactId,
) -> Result<Option<StatusUpdateSerial>> {
let Some(status_update_serial) = self
.write_status_update_inner(&instance.id, &status_update_item, timestamp)
.await?
else {
return Ok(None);
};
let mut notify_msg_id = instance.id;
let mut param_changed = false;
let mut instance = instance.clone();
if let Some(ref document) = status_update_item.document {
if instance
.param
.update_timestamp(Param::WebxdcDocumentTimestamp, timestamp)?
{
instance.param.set(Param::WebxdcDocument, document);
param_changed = true;
}
}
if let Some(ref summary) = status_update_item.summary {
if instance
.param
.update_timestamp(Param::WebxdcSummaryTimestamp, timestamp)?
{
let summary = sanitize_bidi_characters(summary);
instance.param.set(Param::WebxdcSummary, summary.clone());
param_changed = true;
}
}
if can_info_msg {
if let Some(ref info) = status_update_item.info {
let info_msg_id = self
.get_overwritable_info_msg_id(&instance, from_id)
.await?;
if let (Some(info_msg_id), None) = (info_msg_id, &status_update_item.href) {
chat::update_msg_text_and_timestamp(
self,
instance.chat_id,
info_msg_id,
info.as_str(),
timestamp,
)
.await?;
notify_msg_id = info_msg_id;
} else {
notify_msg_id = chat::add_info_msg_with_cmd(
self,
instance.chat_id,
info.as_str(),
SystemMessage::WebxdcInfoMessage,
timestamp,
None,
Some(&instance),
Some(from_id),
)
.await?;
}
if let Some(ref href) = status_update_item.href {
let mut notify_msg = Message::load_from_db(self, notify_msg_id).await?;
notify_msg.param.set(Param::Arg, href);
notify_msg.update_param(self).await?;
}
}
}
if param_changed {
instance.update_param(self).await?;
self.emit_msgs_changed(instance.chat_id, instance.id);
}
if instance.viewtype == Viewtype::Webxdc {
self.emit_event(EventType::WebxdcStatusUpdate {
msg_id: instance.id,
status_update_serial,
});
}
if from_id != ContactId::SELF {
if let Some(notify_list) = status_update_item.notify {
let self_addr = instance.get_webxdc_self_addr(self).await?;
if let Some(notify_text) =
notify_list.get(&self_addr).or_else(|| notify_list.get("*"))
{
self.emit_event(EventType::IncomingWebxdcNotify {
chat_id: instance.chat_id,
contact_id: from_id,
msg_id: notify_msg_id,
text: notify_text.clone(),
href: status_update_item.href,
});
}
}
}
Ok(Some(status_update_serial))
}
pub(crate) async fn write_status_update_inner(
&self,
instance_id: &MsgId,
status_update_item: &StatusUpdateItem,
timestamp: i64,
) -> Result<Option<StatusUpdateSerial>> {
let uid = status_update_item.uid.as_deref();
let status_update_item = serde_json::to_string(&status_update_item)?;
let trans_fn = |t: &mut rusqlite::Transaction| {
t.execute(
"UPDATE msgs SET timestamp_rcvd=? WHERE id=?",
(timestamp, instance_id),
)?;
let rowid = t
.query_row(
"INSERT INTO msgs_status_updates (msg_id, update_item, uid) VALUES(?, ?, ?)
ON CONFLICT (uid) DO NOTHING
RETURNING id",
(instance_id, status_update_item, uid),
|row| {
let id: u32 = row.get(0)?;
Ok(id)
},
)
.optional()?;
Ok(rowid)
};
let Some(rowid) = self.sql.transaction(trans_fn).await? else {
let uid = uid.unwrap_or("-");
info!(self, "Ignoring duplicate status update with uid={uid}");
return Ok(None);
};
let status_update_serial = StatusUpdateSerial(rowid);
Ok(Some(status_update_serial))
}
pub async fn get_status_update(
&self,
msg_id: MsgId,
status_update_serial: StatusUpdateSerial,
) -> Result<String> {
self.sql
.query_get_value(
"SELECT update_item FROM msgs_status_updates WHERE id=? AND msg_id=? ",
(status_update_serial.0, msg_id),
)
.await?
.context("get_status_update: no update item found.")
}
pub async fn send_webxdc_status_update(
&self,
instance_msg_id: MsgId,
update_str: &str,
) -> Result<()> {
let status_update_item: StatusUpdateItem = serde_json::from_str(update_str)
.with_context(|| format!("Failed to parse webxdc update item from {update_str:?}"))?;
self.send_webxdc_status_update_struct(instance_msg_id, status_update_item)
.await?;
Ok(())
}
pub async fn send_webxdc_status_update_struct(
&self,
instance_msg_id: MsgId,
mut status_update: StatusUpdateItem,
) -> Result<()> {
let instance = Message::load_from_db(self, instance_msg_id)
.await
.with_context(|| {
format!("Failed to load message {instance_msg_id} from the database")
})?;
let viewtype = instance.viewtype;
if viewtype != Viewtype::Webxdc {
bail!("send_webxdc_status_update: message {instance_msg_id} is not a webxdc message, but a {viewtype} message.");
}
if instance.param.get_int(Param::WebxdcIntegration).is_some() {
return self
.intercept_send_webxdc_status_update(instance, status_update)
.await;
}
let chat_id = instance.chat_id;
let chat = Chat::load_from_db(self, chat_id)
.await
.with_context(|| format!("Failed to load chat {chat_id} from the database"))?;
if let Some(reason) = chat.why_cant_send(self).await.with_context(|| {
format!("Failed to check if webxdc update can be sent to chat {chat_id}")
})? {
bail!("Cannot send to {chat_id}: {reason}.");
}
let send_now = !matches!(
instance.state,
MessageState::Undefined | MessageState::OutPreparing | MessageState::OutDraft
);
status_update.uid = Some(create_id());
let status_update_serial: StatusUpdateSerial = self
.create_status_update_record(
&instance,
status_update,
create_smeared_timestamp(self),
send_now,
ContactId::SELF,
)
.await
.context("Failed to create status update")?
.context("Duplicate status update UID was generated")?;
if send_now {
self.sql.insert(
"INSERT INTO smtp_status_updates (msg_id, first_serial, last_serial, descr) VALUES(?, ?, ?, '')
ON CONFLICT(msg_id)
DO UPDATE SET last_serial=excluded.last_serial",
(instance.id, status_update_serial, status_update_serial),
).await.context("Failed to insert webxdc update into SMTP queue")?;
self.scheduler.interrupt_smtp().await;
}
Ok(())
}
async fn smtp_status_update_get(&self) -> Result<Option<(MsgId, i64, StatusUpdateSerial)>> {
let res = self
.sql
.query_row_optional(
"SELECT msg_id, first_serial, last_serial \
FROM smtp_status_updates LIMIT 1",
(),
|row| {
let instance_id: MsgId = row.get(0)?;
let first_serial: i64 = row.get(1)?;
let last_serial: StatusUpdateSerial = row.get(2)?;
Ok((instance_id, first_serial, last_serial))
},
)
.await?;
Ok(res)
}
async fn smtp_status_update_pop_serials(
&self,
msg_id: MsgId,
first: i64,
first_new: StatusUpdateSerial,
) -> Result<()> {
if self
.sql
.execute(
"DELETE FROM smtp_status_updates \
WHERE msg_id=? AND first_serial=? AND last_serial<?",
(msg_id, first, first_new),
)
.await?
> 0
{
return Ok(());
}
self.sql
.execute(
"UPDATE smtp_status_updates SET first_serial=? \
WHERE msg_id=? AND first_serial=?",
(first_new, msg_id, first),
)
.await?;
Ok(())
}
pub(crate) async fn flush_status_updates(&self) -> Result<()> {
loop {
let (instance_id, first, last) = match self.smtp_status_update_get().await? {
Some(res) => res,
None => return Ok(()),
};
let (json, first_new) = self
.render_webxdc_status_update_object(
instance_id,
StatusUpdateSerial(max(first, 1).try_into()?),
last,
Some(STATUS_UPDATE_SIZE_MAX),
)
.await?;
if let Some(json) = json {
let instance = Message::load_from_db(self, instance_id).await?;
let mut status_update = Message {
chat_id: instance.chat_id,
viewtype: Viewtype::Text,
text: BODY_DESCR.to_string(),
hidden: true,
..Default::default()
};
status_update
.param
.set_cmd(SystemMessage::WebxdcStatusUpdate);
status_update.param.set(Param::Arg, json);
status_update.set_quote(self, Some(&instance)).await?;
status_update.param.remove(Param::GuaranteeE2ee); chat::send_msg(self, instance.chat_id, &mut status_update).await?;
}
self.smtp_status_update_pop_serials(instance_id, first, first_new)
.await?;
}
}
pub(crate) fn build_status_update_part(&self, json: &str) -> MimePart<'static> {
MimePart::new("application/json", json.as_bytes().to_vec()).attachment("status-update.json")
}
pub(crate) async fn receive_status_update(
&self,
from_id: ContactId,
instance: &Message,
timestamp: i64,
can_info_msg: bool,
json: &str,
) -> Result<()> {
let chat_id = instance.chat_id;
if from_id != ContactId::SELF && !chat::is_contact_in_chat(self, chat_id, from_id).await? {
let chat_type: Chattype = self
.sql
.query_get_value("SELECT type FROM chats WHERE id=?", (chat_id,))
.await?
.with_context(|| format!("Chat type for chat {chat_id} not found"))?;
if chat_type != Chattype::Mailinglist {
bail!("receive_status_update: status sender {from_id} is not a member of chat {chat_id}")
}
}
let updates: StatusUpdates = serde_json::from_str(json)?;
for update_item in updates.updates {
self.create_status_update_record(
instance,
update_item,
timestamp,
can_info_msg,
from_id,
)
.await?;
}
Ok(())
}
pub async fn get_webxdc_status_updates(
&self,
instance_msg_id: MsgId,
last_known_serial: StatusUpdateSerial,
) -> Result<String> {
let param = instance_msg_id.get_param(self).await?;
if param.get_int(Param::WebxdcIntegration).is_some() {
let instance = Message::load_from_db(self, instance_msg_id).await?;
return self
.intercept_get_webxdc_status_updates(instance, last_known_serial)
.await;
}
let json = self
.sql
.query_map(
"SELECT update_item, id FROM msgs_status_updates WHERE msg_id=? AND id>? ORDER BY id",
(instance_msg_id, last_known_serial),
|row| {
let update_item_str = row.get::<_, String>(0)?;
let serial = row.get::<_, StatusUpdateSerial>(1)?;
Ok((update_item_str, serial))
},
|rows| {
let mut rows_copy : Vec<(String, StatusUpdateSerial)> = Vec::new(); let mut max_serial = StatusUpdateSerial(0);
for row in rows {
let row = row?;
if row.1 > max_serial {
max_serial = row.1;
}
rows_copy.push(row);
}
let mut json = String::default();
for row in rows_copy {
let (update_item_str, serial) = row;
let update_item = StatusUpdateItemAndSerial
{
item: StatusUpdateItem {
uid: None, ..serde_json::from_str(&update_item_str)?
},
serial,
max_serial,
};
if !json.is_empty() {
json.push_str(",\n");
}
json.push_str(&serde_json::to_string(&update_item)?);
}
Ok(json)
},
)
.await?;
Ok(format!("[{json}]"))
}
pub(crate) async fn render_webxdc_status_update_object(
&self,
instance_msg_id: MsgId,
first: StatusUpdateSerial,
last: StatusUpdateSerial,
size_max: Option<usize>,
) -> Result<(Option<String>, StatusUpdateSerial)> {
let (json, first_new) = self
.sql
.query_map(
"SELECT id, update_item FROM msgs_status_updates \
WHERE msg_id=? AND id>=? AND id<=? ORDER BY id",
(instance_msg_id, first, last),
|row| {
let id: StatusUpdateSerial = row.get(0)?;
let update_item: String = row.get(1)?;
Ok((id, update_item))
},
|rows| {
let mut json = String::default();
for row in rows {
let (id, update_item) = row?;
if !json.is_empty()
&& json.len() + update_item.len() >= size_max.unwrap_or(usize::MAX)
{
return Ok((json, id));
}
if !json.is_empty() {
json.push_str(",\n");
}
json.push_str(&update_item);
}
Ok((
json,
StatusUpdateSerial::new(last.to_u32().saturating_add(1)),
))
},
)
.await?;
let json = match json.is_empty() {
true => None,
false => Some(format!(r#"{{"updates":[{json}]}}"#)),
};
Ok((json, first_new))
}
}
fn parse_webxdc_manifest(bytes: &[u8]) -> Result<WebxdcManifest> {
let s = std::str::from_utf8(bytes)?;
let manifest: WebxdcManifest = toml::from_str(s)?;
Ok(manifest)
}
async fn get_blob(archive: &mut SeekZipFileReader<BufReader<File>>, name: &str) -> Result<Vec<u8>> {
let (i, _) = find_zip_entry(archive.file(), name)
.ok_or_else(|| anyhow!("no entry found for {}", name))?;
let mut reader = archive.reader_with_entry(i).await?;
let mut buf = Vec::new();
reader.read_to_end_checked(&mut buf).await?;
Ok(buf)
}
impl Message {
async fn get_webxdc_archive(
&self,
context: &Context,
) -> Result<SeekZipFileReader<BufReader<File>>> {
let path = self
.get_file(context)
.ok_or_else(|| format_err!("No webxdc instance file."))?;
let path_abs = get_abs_path(context, &path);
let file = BufReader::new(File::open(path_abs).await?);
let archive = SeekZipFileReader::with_tokio(file).await?;
Ok(archive)
}
pub async fn get_webxdc_blob(&self, context: &Context, name: &str) -> Result<Vec<u8>> {
ensure!(self.viewtype == Viewtype::Webxdc, "No webxdc instance.");
if name == WEBXDC_DEFAULT_ICON {
return Ok(include_bytes!("../assets/icon-webxdc.png").to_vec());
}
let name = if name.starts_with('/') {
name.split_at(1).1
} else {
name
};
let mut archive = self.get_webxdc_archive(context).await?;
if name == "index.html" {
if let Ok(bytes) = get_blob(&mut archive, "manifest.toml").await {
if let Ok(manifest) = parse_webxdc_manifest(&bytes) {
if let Some(min_api) = manifest.min_api {
if min_api > WEBXDC_API_VERSION {
return Ok(Vec::from(
"<!DOCTYPE html>This Webxdc requires a newer Delta Chat version.",
));
}
}
}
}
}
get_blob(&mut archive, name).await
}
pub async fn get_webxdc_info(&self, context: &Context) -> Result<WebxdcInfo> {
ensure!(self.viewtype == Viewtype::Webxdc, "No webxdc instance.");
let mut archive = self.get_webxdc_archive(context).await?;
let mut manifest = get_blob(&mut archive, "manifest.toml")
.await
.map(|bytes| parse_webxdc_manifest(&bytes).unwrap_or_default())
.unwrap_or_default();
if let Some(ref name) = manifest.name {
let name = name.trim();
if name.is_empty() {
warn!(context, "empty name given in manifest");
manifest.name = None;
}
}
let request_integration = manifest.request_integration.unwrap_or_default();
let is_integrated = self.is_set_as_webxdc_integration(context).await?;
let internet_access = is_integrated;
let self_addr = self.get_webxdc_self_addr(context).await?;
Ok(WebxdcInfo {
name: if let Some(name) = manifest.name {
name
} else {
self.get_filename().unwrap_or_default()
},
icon: if find_zip_entry(archive.file(), "icon.png").is_some() {
"icon.png".to_string()
} else if find_zip_entry(archive.file(), "icon.jpg").is_some() {
"icon.jpg".to_string()
} else {
WEBXDC_DEFAULT_ICON.to_string()
},
document: self
.param
.get(Param::WebxdcDocument)
.unwrap_or_default()
.to_string(),
summary: if is_integrated {
"🌍 Used as map. Delete to use default. Do not enter sensitive data".to_string()
} else if request_integration == "map" {
"🌏 To use as map, forward to \"Saved Messages\" again. Do not enter sensitive data"
.to_string()
} else {
self.param
.get(Param::WebxdcSummary)
.unwrap_or_default()
.to_string()
},
source_code_url: if let Some(url) = manifest.source_code_url {
url
} else {
"".to_string()
},
request_integration,
internet_access,
self_addr,
send_update_interval: context.ratelimit.read().await.update_interval(),
send_update_max_size: RECOMMENDED_FILE_SIZE as usize,
})
}
async fn get_webxdc_self_addr(&self, context: &Context) -> Result<String> {
let fingerprint = load_self_public_key(context).await?.dc_fingerprint().hex();
let data = format!("{}-{}", fingerprint, self.rfc724_mid);
let hash = Sha256::digest(data.as_bytes());
Ok(format!("{:x}", hash))
}
pub fn get_webxdc_href(&self) -> Option<String> {
self.param.get(Param::Arg).map(|href| href.to_string())
}
}
#[cfg(test)]
mod webxdc_tests;