use std::path::Path;
use anyhow::{Context, Result, anyhow};
use jiff::Timestamp;
use log::{debug, error, warn};
use pyo3::prelude::*;
use pyo3::types::{IntoPyDict, PyCFunction, PyDict, PyTuple};
use serde::Serialize;
use sha2::{Digest, Sha256};
use crate::rns::{PyRNS, PyRNSDestination, PyRNSDestinationHash, PyRNSIdentity, RNSDestination};
use crate::utils::{LongHash, ShortHash};
#[derive(Debug)]
struct PyLXMessage(PyObject);
#[derive(Debug, Clone)]
pub struct LXMessage {
pub title: String,
pub content: String,
pub fields: rmpv::Value,
pub timestamp: Timestamp,
pub destination_hash: ShortHash,
pub source_hash: ShortHash,
pub state: LXMessageState,
pub desired_method: Option<LXMessageMethod>,
pub method: Option<LXMessageMethod>,
}
impl LXMessage {
pub fn new(
title: String,
content: String,
timestamp: Timestamp,
destination_hash: ShortHash,
source_hash: ShortHash,
desired_method: LXMessageMethod,
) -> Self {
Self {
title,
content,
fields: rmpv::Value::Nil,
timestamp,
destination_hash,
source_hash,
state: LXMessageState::GENERATING,
desired_method: Some(desired_method),
method: None,
}
}
fn from_raw(raw_msg: LXMessageRaw) -> Result<Self> {
let title = String::from_utf8_lossy(&raw_msg.title).into_owned();
let content = String::from_utf8_lossy(&raw_msg.content).into_owned();
let ts = (raw_msg.timestamp * 1_000_000_000_f64) as i128;
let timestamp = Timestamp::from_nanosecond(ts).unwrap_or_default();
let fields_msgpacked: Result<Vec<u8>> = Python::with_gil(|py| {
let msgpacker = PyModule::import(py, "RNS.vendor.umsgpack")?;
let msgpacked: Result<Vec<u8>> = msgpacker
.call_method1("dumps", (raw_msg.fields,))
.context("failed to have python convert fields to json")
.and_then(|p| p.extract().context("Failed to extract python value"))
.inspect_err(|e| warn!("failed to extract fields as json:\n{e:?}"));
msgpacked
});
let fields = fields_msgpacked
.and_then(|v| {
rmpv::decode::read_value(&mut v.as_slice())
.context("failed to decode msgpacked fields data")
})
.inspect_err(|e| warn!("{e:?}"))
.unwrap_or(rmpv::Value::Nil);
let state: LXMessageState = raw_msg.state.try_into()?;
let desired_method: Option<LXMessageMethod> = raw_msg
.desired_method
.map(LXMessageMethod::try_from)
.transpose()?;
let method: Option<LXMessageMethod> =
raw_msg.method.map(LXMessageMethod::try_from).transpose()?;
let v = Self {
title,
content,
fields,
timestamp,
destination_hash: raw_msg.destination_hash,
source_hash: raw_msg.source_hash,
state,
desired_method,
method,
};
Ok(v)
}
fn pack_payload(&self) -> Result<Vec<u8>> {
let timestamp = (self.timestamp.as_nanosecond() as f64) / 1_000_000_000_f64;
let timestamp = rmpv::Value::F64(timestamp);
let title = rmpv::Value::Binary(self.title.clone().into_bytes());
let content = rmpv::Value::Binary(self.content.clone().into_bytes());
let mut fields_value = vec![];
let mut value_serializer = rmp_serde::Serializer::new(&mut fields_value);
self.fields.serialize(&mut value_serializer)?;
let fields_value: rmpv::Value = rmp_serde::from_slice(&fields_value)?;
let fields = fields_value;
let payload = [timestamp, title, content, fields];
let payload = rmp_serde::to_vec(&payload)?;
Ok(payload)
}
fn hashed_part(&self) -> Result<Vec<u8>> {
let destination_hash = self.destination_hash.value();
let source_hash = self.source_hash.value();
let mut payload = self.pack_payload()?;
let mut hashed_part: Vec<u8> = vec![];
hashed_part.extend(destination_hash);
hashed_part.extend(source_hash);
hashed_part.append(&mut payload);
Ok(hashed_part)
}
pub fn message_id(&self) -> Result<LongHash> {
let hashed = Sha256::digest(self.hashed_part()?);
let long_hash = LongHash::new(hashed.as_slice())?;
Ok(long_hash)
}
}
#[derive(Debug, FromPyObject)]
struct LXMessageRaw {
title: Vec<u8>,
content: Vec<u8>,
fields: PyObject,
timestamp: f64,
destination_hash: ShortHash,
source_hash: ShortHash,
state: u8,
desired_method: Option<u8>,
method: Option<u8>,
}
struct PyLXMFRouter(PyObject);
pub struct PyLXMF {
router: PyLXMFRouter,
our_destination: RNSDestination,
our_destination_hash: PyRNSDestinationHash,
}
impl PyLXMF {
pub fn new(rns: &PyRNS, display_name: &str, storage_path: &Path) -> Result<Self> {
Python::with_gil(|py| {
let lxmf_mod = PyModule::import(py, "LXMF")?;
let lxmf_router_kwargs =
[("storagepath", storage_path.to_string_lossy())].into_py_dict(py)?;
let lxmf_router = lxmf_mod.call_method("LXMRouter", (), Some(&lxmf_router_kwargs))?;
let register_delivery_entity_kwargs =
[("display_name", display_name)].into_py_dict(py)?;
let my_lxmf_destination = lxmf_router.call_method(
"register_delivery_identity",
(rns.identity.0.bind(py),),
Some(®ister_delivery_entity_kwargs),
)?;
let my_lxmf_destination_hash = my_lxmf_destination.getattr("hash")?;
Ok(Self {
router: PyLXMFRouter(lxmf_router.unbind()),
our_destination_hash: PyRNSDestinationHash(my_lxmf_destination_hash.extract()?),
our_destination: RNSDestination(PyRNSDestination(my_lxmf_destination.unbind())),
})
})
}
pub fn set_callback<F>(&self, callback: F) -> Result<()>
where
F: Fn(LXMessage) + Send + 'static,
{
Python::with_gil(|py| {
let callback_wrapper = wrap_callback(callback)?;
let kwargs = [("callback", callback_wrapper)].into_py_dict(py)?;
self.router
.0
.bind(py)
.call_method("register_delivery_callback", (), Some(&kwargs))?;
Ok(())
})
}
pub fn announce(&self) -> Result<()> {
Python::with_gil(|py| {
let lxmf_router = &self.router.0;
let my_lxmf_destination_hash = self.our_destination_hash.0.value();
lxmf_router.call_method1(py, "announce", (&my_lxmf_destination_hash,))?;
Ok(())
})
}
fn create_lxmessage_py<D, F>(
&self,
message: &LXMessage,
delivery_method: LXMessageMethod,
destination: &RNSDestination,
delivery_callback: Option<D>,
failed_callback: Option<F>,
) -> Result<PyLXMessage>
where
D: Fn(LXMessage) + Send + 'static,
F: Fn(LXMessage) + Send + 'static,
{
Python::with_gil(|py| {
let lxmf_mod = PyModule::import(py, "LXMF")?;
let destination = &destination.0.0;
let source = &self.our_destination.0.0;
let content = &message.content;
let title = &message.title;
let include_ticket = true;
let kwargs = PyDict::new(py);
kwargs.set_item("destination", destination)?;
kwargs.set_item("source", source)?;
kwargs.set_item("content", content)?;
kwargs.set_item("title", title)?;
kwargs.set_item("desired_method", delivery_method as isize)?;
kwargs.set_item("stamp_cost", None::<()>)?;
kwargs.set_item("include_ticket", include_ticket)?;
let lxm = lxmf_mod.call_method("LXMessage", (), Some(&kwargs))?;
let ts_ns = message.timestamp.as_nanosecond() as f64 / 1_000_000_000.00;
lxm.setattr("timestamp", ts_ns)?;
if let Some(delivery_callback) = delivery_callback {
let callback_wrapper = wrap_callback(delivery_callback)?;
lxm.call_method1("register_delivery_callback", (callback_wrapper,))?;
debug!(
"adding 'delivery' callback to message {}",
message.message_id()?
);
}
if let Some(failed_callback) = failed_callback {
let callback_wrapper = wrap_callback(failed_callback)?;
lxm.call_method1("register_failed_callback", (callback_wrapper,))?;
debug!(
"adding 'failed' callback to message {}",
message.message_id()?
)
}
Ok(PyLXMessage(lxm.unbind()))
})
}
pub fn send(&self, message: LXMessage, destination: &RNSDestination) -> Result<()> {
self.send_with_callbacks::<fn(LXMessage), fn(LXMessage)>(message, destination, None, None)
}
pub fn send_with_callbacks<D, F>(
&self,
message: LXMessage,
destination: &RNSDestination,
delivery_callback: Option<D>,
failed_callback: Option<F>,
) -> Result<()>
where
D: Fn(LXMessage) + Send + 'static,
F: Fn(LXMessage) + Send + 'static,
{
Python::with_gil(|py| {
let router = &self.router.0;
let lxm = self.create_lxmessage_py(
&message,
message.desired_method.unwrap_or_default(),
destination,
delivery_callback,
failed_callback,
)?;
router.call_method1(py, "handle_outbound", (lxm.0,))?;
debug!(
"sent a message to {}: {:?} - {:?}",
message.destination_hash, message.title, message.content
);
Ok(())
})
}
pub fn lxmf_destination_hash(&self) -> &ShortHash {
&self.our_destination_hash.0
}
pub fn display_name_from_app_data(&self, app_data: &[u8]) -> Result<Option<String>> {
Python::with_gil(|py| {
let lxmf_mod = PyModule::import(py, "LXMF")?;
let display_name: Option<String> = lxmf_mod
.call_method1("display_name_from_app_data", (app_data,))?
.extract()?;
Ok(display_name)
})
}
pub fn set_propagation_node(&self, propagation_node_hash: ShortHash) -> Result<()> {
Python::with_gil(|py| {
self.router.0.call_method1(
py,
"set_active_propagation_node",
(propagation_node_hash.value(),),
)?;
Ok(())
})
}
pub fn request_messages_from_propagation_node(&self, identity: &PyRNSIdentity) -> Result<()> {
Python::with_gil(|py| {
self.router.0.call_method1(
py,
"request_messages_from_propagation_node",
(&identity.0,),
)?;
Ok(())
})
}
}
fn wrap_callback<F>(callback: F) -> Result<Py<PyCFunction>>
where
F: Fn(LXMessage) + Send + 'static,
{
Python::with_gil(|py| {
let callback_wrapper =
move |args: &Bound<'_, PyTuple>, _kwargs: Option<&Bound<'_, PyDict>>| {
let payload = args.get_item(0).expect("failed to get first item of tuple");
let raw_msg = payload.extract::<LXMessageRaw>();
match raw_msg {
Ok(msg) => {
let parsed_msg = LXMessage::from_raw(msg);
match parsed_msg {
Ok(m) => callback(m),
Err(e) => error!("further parsing errored: {e:?}"),
}
}
Err(err) => error!("parsing errored: {err:?}"),
}
};
let some_callback = PyCFunction::new_closure(py, None, None, callback_wrapper)?.unbind();
Ok(some_callback)
})
}
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub enum LXMessageMethod {
OPPORTUNISTIC = 0x01,
#[default]
DIRECT = 0x02,
PROPAGATED = 0x03,
PAPER = 0x05,
}
impl TryFrom<u8> for LXMessageMethod {
type Error = anyhow::Error;
fn try_from(value: u8) -> std::result::Result<Self, Self::Error> {
match value {
1 => Ok(Self::OPPORTUNISTIC),
2 => Ok(Self::DIRECT),
3 => Ok(Self::PROPAGATED),
5 => Ok(Self::PAPER),
_ => Err(anyhow!(
"Failed to parse LXMessage Method: unknown variant '{value}'"
)),
}
}
}
#[derive(Debug, Clone)]
pub enum LXMessageState {
GENERATING = 0x00,
OUTBOUND = 0x01,
SENDING = 0x02,
SENT = 0x04,
DELIVERED = 0x08,
REJECTED = 0xFD,
CANCELLED = 0xFE,
FAILED = 0xFF,
}
impl TryFrom<u8> for LXMessageState {
type Error = anyhow::Error;
fn try_from(value: u8) -> std::result::Result<Self, Self::Error> {
match value {
0 => Ok(Self::GENERATING),
1 => Ok(Self::OUTBOUND),
2 => Ok(Self::SENDING),
4 => Ok(Self::SENT),
8 => Ok(Self::DELIVERED),
253 => Ok(Self::REJECTED),
254 => Ok(Self::CANCELLED),
255 => Ok(Self::FAILED),
_ => Err(anyhow!(
"Failed to parse LXMessage State: unknown variant '{value}'"
)),
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::rns::PrivateKey;
use hex_literal::hex;
use pyo3_ffi::c_str;
fn _can_calculate_message_id() {
let identity = include_bytes!("../tests/some_identity");
let identity = PrivateKey::from_bytes(identity.to_vec());
let rns = PyRNS::initialize(Some(identity)).unwrap();
let lxmf = PyLXMF::new(&rns, "foo", Path::new("/tmp/lxmf_test_store")).unwrap();
let some_destination = &lxmf.our_destination.0.0;
let (lxmessage, python_hash) = Python::with_gil(|py| {
let lxmf_mod = PyModule::import(py, "LXMF").unwrap();
let content = "content".to_string();
let title = "title".to_string();
let include_ticket = true;
let kwargs = PyDict::new(py);
kwargs.set_item("destination", some_destination).unwrap();
kwargs.set_item("source", some_destination).unwrap();
kwargs.set_item("content", &content).unwrap();
kwargs.set_item("title", &title).unwrap();
kwargs
.set_item("desired_method", LXMessageMethod::DIRECT as isize)
.unwrap();
kwargs.set_item("stamp_cost", None::<()>).unwrap();
kwargs.set_item("include_ticket", include_ticket).unwrap();
let lxmessage = lxmf_mod
.call_method("LXMessage", (), Some(&kwargs))
.unwrap();
lxmessage
.setattr("timestamp", 1754646238.4003625_f64)
.unwrap();
lxmessage.call_method0("pack").unwrap();
let msg_id: LongHash = lxmessage.getattr("message_id").unwrap().extract().unwrap();
let raw_msg: LXMessageRaw = lxmessage.extract().unwrap();
(LXMessage::from_raw(raw_msg).unwrap(), msg_id)
});
assert_eq!(
hex!("8007e714fcc4de9d488949eac2df04142ed408748f4aef91870560d36e6d0bd5"),
python_hash.value(),
);
assert_eq!(python_hash, lxmessage.message_id().unwrap());
}
fn _test_hash_calcs() {
let identity = include_bytes!("../tests/some_identity");
let identity = PrivateKey::from_bytes(identity.to_vec());
let rns = PyRNS::initialize(Some(identity)).unwrap();
let lxmf = PyLXMF::new(&rns, "foo", Path::new("/tmp/lxmf_test_store")).unwrap();
let some_destination = &lxmf.our_destination.0.0;
Python::with_gil(|py| {
let lxmf_mod = PyModule::import(py, "LXMF").unwrap();
let content = "content".to_string();
let title = "title".to_string();
let include_ticket = true;
let kwargs = PyDict::new(py);
kwargs.set_item("destination", some_destination).unwrap();
kwargs.set_item("source", some_destination).unwrap();
kwargs.set_item("content", &content).unwrap();
kwargs.set_item("title", &title).unwrap();
kwargs
.set_item("desired_method", LXMessageMethod::DIRECT as isize)
.unwrap();
kwargs.set_item("stamp_cost", None::<()>).unwrap();
kwargs.set_item("include_ticket", include_ticket).unwrap();
let lxmessage = lxmf_mod
.call_method("LXMessage", (), Some(&kwargs))
.unwrap();
lxmessage
.setattr("timestamp", 1754646238.4003625_f64)
.unwrap();
lxmessage.call_method0("pack").unwrap();
let timestamp: f64 = lxmessage.getattr("timestamp").unwrap().extract().unwrap();
let title: Vec<u8> = lxmessage.getattr("title").unwrap().extract().unwrap();
let content: Vec<u8> = lxmessage.getattr("content").unwrap().extract().unwrap();
let raw_msg: LXMessageRaw = lxmessage.extract().unwrap();
assert_eq!(timestamp, raw_msg.timestamp);
assert_eq!(title, raw_msg.title);
assert_eq!(content, raw_msg.content);
let msg = LXMessage::from_raw(raw_msg).unwrap();
let hashiness = PyModule::from_code(
py,
c_str!(
r#"
import RNS.vendor.umsgpack as msgpack
import RNS
def dest_hashes(x):
hashed_part = b""
hashed_part += x.destination.hash
hashed_part += x.source.hash
return hashed_part
def part_payload_hash(x):
part_payload = [x.timestamp, x.title, x.content, x.fields]
return msgpack.packb(part_payload)
def hashed_part(x):
payload = [x.timestamp, x.title, x.content, x.fields]
hashed_part = b""
hashed_part += x.destination.hash
hashed_part += x.source.hash
hashed_part += msgpack.packb(payload)
return hashed_part
def message_id(x):
payload = [x.timestamp, x.title, x.content, x.fields]
hashed_part = b""
hashed_part += x.destination.hash
hashed_part += x.source.hash
hashed_part += msgpack.packb(payload)
message_id = RNS.Identity.full_hash(hashed_part)
return message_id
def hash_thing(x):
return RNS.Identity.full_hash(x)
"#
),
c_str!("hashiness.py"),
c_str!("hashiness"),
)
.unwrap();
let hashed_dest_source_py: Vec<u8> = hashiness
.getattr("dest_hashes")
.unwrap()
.call1((lxmessage.clone(),))
.unwrap()
.extract()
.unwrap();
let packed_part_payload_py: Vec<u8> = hashiness
.getattr("part_payload_hash")
.unwrap()
.call1((lxmessage.clone(),))
.unwrap()
.extract()
.unwrap();
let message_id_py: Vec<u8> = hashiness
.getattr("message_id")
.unwrap()
.call1((lxmessage.clone(),))
.unwrap()
.extract()
.unwrap();
let message_id_py_proper: Vec<u8> =
lxmessage.getattr("message_id").unwrap().extract().unwrap();
let message_id_py_proper_lh: LongHash =
lxmessage.getattr("message_id").unwrap().extract().unwrap();
assert_eq!(message_id_py, message_id_py_proper);
assert_eq!(message_id_py, message_id_py_proper_lh.value());
let mut hashed_dest_source_rs: Vec<u8> = vec![];
hashed_dest_source_rs.extend(msg.destination_hash.value());
hashed_dest_source_rs.extend(msg.source_hash.value());
assert_eq!(hashed_dest_source_py, hashed_dest_source_rs);
let rust_payload = msg.pack_payload().unwrap();
assert_eq!(
timestamp,
(msg.timestamp.as_nanosecond() as f64 / 1_000_000_000.00)
);
assert_eq!(packed_part_payload_py, rust_payload);
let hashed_part_py: Vec<u8> = hashiness
.getattr("hashed_part")
.unwrap()
.call1((lxmessage,))
.unwrap()
.extract()
.unwrap();
let hashed_part_rs = msg.hashed_part().unwrap();
assert_eq!(hashed_part_py, hashed_part_rs);
println!(
"py: {}\nrs: {}",
hex::encode(hashed_part_py.clone()),
hex::encode(hashed_part_rs.clone())
);
let rust_coded_hashed_py = Sha256::digest(&hashed_part_py);
let rust_coded_hashed_rs = Sha256::digest(&hashed_part_rs);
let python_coded_hashed_py: Vec<u8> = hashiness
.getattr("hash_thing")
.unwrap()
.call1((hashed_part_py,))
.unwrap()
.extract()
.unwrap();
let python_coded_hashed_rs: Vec<u8> = hashiness
.getattr("hash_thing")
.unwrap()
.call1((hashed_part_rs,))
.unwrap()
.extract()
.unwrap();
assert_eq!(rust_coded_hashed_py, rust_coded_hashed_rs);
assert_eq!(rust_coded_hashed_py.as_slice(), python_coded_hashed_py);
assert_eq!(rust_coded_hashed_py.as_slice(), python_coded_hashed_rs);
let message_id_rs = msg.message_id().unwrap();
println!(
"py: {}\nrs: {}",
hex::encode(&message_id_py),
hex::encode(message_id_rs.value()),
);
assert_eq!(message_id_py, message_id_rs.value());
});
}
}