xmppftw/xmpp-rs
Watch
1
0
Fork
You've already forked xmpp-rs
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Merge branch 'lm-master'

Browse source 
Merging xmpp-rs/xmpp-rs and linkmauve/xmpp-rs.

Lots has been happening in our small ecosystem, this is our Nth attempt at
finding an API that we like. There might still be issues with this one but it
looks good enough so that we can start using it for relatively simple clients.
If it happens that the API is problematic then we'll change again. S%#$
happens.

With this merge, the focus of the library shits a bit.

This library is aimed to be a high-level library and provide an API somewhat
abstracted from XMPP.

We are also now using tokio-xmpp as the underlying library managing the stream,
and not doing it ourselves (even though nothing technically prevents it).
This commit is contained in:
Maxime “pep” Buquet 2019-09-13 01:08:27 +02:00
commit 289437d5b3
36 changed files with 952 additions and 3582 deletions
Download patch file Download diff file Expand all files Collapse all files

97
src/avatar.rs Normal file
View file

@ -0,0 +1,97 @@
// Copyright (c) 2019 Emmanuel Gil Peyrot <linkmauve@linkmauve.fr>
//
// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at http://mozilla.org/MPL/2.0/.
use crate::Event;
use futures::{sync::mpsc, Sink};
use std::convert::TryFrom;
use std::fs::{self, File};
use std::io::{self, Write};
use tokio_xmpp::Packet;
use xmpp_parsers::{
avatar::{Data, Metadata},
iq::Iq,
ns,
pubsub::{
event::Item,
pubsub::{Items, PubSub},
NodeName,
},
hashes::Hash,
Jid,
};
// TODO: Update xmpp-parsers to get this function for free on Hash.
fn hash_to_hex(hash: &Hash) -> String {
let mut bytes = vec![];
for byte in hash.hash.iter() {
bytes.push(format!("{:02x}", byte));
}
bytes.join("")
}
pub(crate) fn handle_metadata_pubsub_event(from: &Jid, tx: &mut mpsc::UnboundedSender<Packet>, items: Vec<Item>) -> impl IntoIterator<Item = Event> {
let mut events = Vec::new();
for item in items {
let payload = item.payload.clone().unwrap();
if payload.is("metadata", ns::AVATAR_METADATA) {
let metadata = Metadata::try_from(payload).unwrap();
for info in metadata.infos {
let filename = format!("data/{}/{}", from, hash_to_hex(&*info.id));
let file_length = match fs::metadata(filename.clone()) {
Ok(metadata) => metadata.len(),
Err(_) => 0,
};
// TODO: Also check the hash.
if info.bytes as u64 == file_length {
events.push(Event::AvatarRetrieved(from.clone(), filename));
} else {
let iq = download_avatar(from);
tx.start_send(Packet::Stanza(iq.into())).unwrap();
}
}
}
}
events
}
fn download_avatar(from: &Jid) -> Iq {
Iq::from_get("coucou", PubSub::Items(Items {
max_items: None,
node: NodeName(String::from(ns::AVATAR_DATA)),
subid: None,
items: Vec::new(),
}))
.with_to(from.clone())
}
// The return value of this function will be simply pushed to a Vec in the caller function,
// so it makes no sense to allocate a Vec here - we're lazy instead
pub(crate) fn handle_data_pubsub_iq<'a>(
from: &'a Jid,
items: &'a Items,
) -> impl IntoIterator<Item = Event> + 'a {
let from = from.clone();
items
.items
.iter()
.filter_map(move |item| match (&item.id, &item.payload) {
(Some(id), Some(payload)) => {
let data = Data::try_from(payload.clone()).unwrap();
let filename = save_avatar(&from, id.0.clone(), &data.data).unwrap();
Some(Event::AvatarRetrieved(from.clone(), filename))
}
_ => None,
})
}
fn save_avatar(from: &Jid, id: String, data: &[u8]) -> io::Result<String> {
let directory = format!("data/{}", from);
let filename = format!("data/{}/{}", from, id);
fs::create_dir_all(directory)?;
let mut file = File::create(&filename)?;
file.write_all(data)?;
Ok(filename)
}

297
src/client.rs
View file

@ -1,297 +0,0 @@
use xml;
use jid::Jid;
use transport::{Transport, SslTransport};
use error::Error;
use ns;
use plugin::{Plugin, PluginInit, PluginProxyBinding, PluginContainer, PluginRef};
use connection::{Connection, C2S};
use sasl::client::Mechanism as SaslMechanism;
use sasl::client::mechanisms::{Plain, Scram};
use sasl::common::{Credentials as SaslCredentials, Identity, Secret, ChannelBinding};
use sasl::common::scram::{Sha1, Sha256};
use components::sasl_error::SaslError;
use util::FromElement;
use event::{Event, Dispatcher, Propagation, SendElement, ReceiveElement, Priority};
use base64;
use minidom::Element;
use xml::reader::XmlEvent as ReaderEvent;
use std::sync::{Mutex, Arc};
use std::collections::HashSet;
/// Struct that should be moved somewhere else and cleaned up.
#[derive(Debug)]
pub struct StreamFeatures {
pub sasl_mechanisms: Option<HashSet<String>>,
}
/// A builder for `Client`s.
pub struct ClientBuilder {
jid: Jid,
credentials: SaslCredentials,
host: Option<String>,
port: u16,
}
impl ClientBuilder {
/// Creates a new builder for an XMPP client that will connect to `jid` with default parameters.
pub fn new(jid: Jid) -> ClientBuilder {
ClientBuilder {
jid: jid,
credentials: SaslCredentials::default(),
host: None,
port: 5222,
}
}
/// Sets the host to connect to.
pub fn host(mut self, host: String) -> ClientBuilder {
self.host = Some(host);
self
}
/// Sets the port to connect to.
pub fn port(mut self, port: u16) -> ClientBuilder {
self.port = port;
self
}
/// Sets the password to use.
pub fn password<P: Into<String>>(mut self, password: P) -> ClientBuilder {
self.credentials = SaslCredentials {
identity: Identity::Username(self.jid.node.clone().expect("JID has no node")),
secret: Secret::password_plain(password),
channel_binding: ChannelBinding::None,
};
self
}
/// Connects to the server and returns a `Client` when succesful.
pub fn connect(self) -> Result<Client, Error> {
let host = &self.host.unwrap_or(self.jid.domain.clone());
let mut transport = SslTransport::connect(host, self.port)?;
C2S::init(&mut transport, &self.jid.domain, "before_sasl")?;
let dispatcher = Arc::new(Dispatcher::new());
let mut credentials = self.credentials;
credentials.channel_binding = transport.channel_bind();
let transport = Arc::new(Mutex::new(transport));
let plugin_container = Arc::new(PluginContainer::new());
let mut client = Client {
jid: self.jid.clone(),
transport: transport.clone(),
binding: PluginProxyBinding::new(dispatcher.clone(), plugin_container.clone(), self.jid),
plugin_container: plugin_container,
dispatcher: dispatcher,
};
client.dispatcher.register(Priority::Default, move |evt: &SendElement| {
let mut t = transport.lock().unwrap();
t.write_element(&evt.0).unwrap();
Propagation::Continue
});
client.connect(credentials)?;
client.bind()?;
Ok(client)
}
}
/// An XMPP client.
pub struct Client {
jid: Jid,
transport: Arc<Mutex<SslTransport>>,
plugin_container: Arc<PluginContainer>,
binding: PluginProxyBinding,
dispatcher: Arc<Dispatcher>,
}
impl Client {
/// Returns a reference to the `Jid` associated with this `Client`.
pub fn jid(&self) -> &Jid {
&self.jid
}
/// Registers a plugin.
pub fn register_plugin<P: Plugin + PluginInit + 'static>(&mut self, mut plugin: P) {
let binding = self.binding.clone();
plugin.bind(binding);
let p = Arc::new(plugin);
P::init(&self.dispatcher, p.clone());
self.plugin_container.register(p);
}
pub fn register_handler<E, F>(&mut self, pri: Priority, func: F)
where
E: Event,
F: Fn(&E) -> Propagation + 'static {
self.dispatcher.register(pri, func);
}
/// Returns the plugin given by the type parameter, if it exists, else panics.
pub fn plugin<P: Plugin>(&self) -> PluginRef<P> {
self.plugin_container.get::<P>().unwrap()
}
/// Returns the next event and flush the send queue.
pub fn main(&mut self) -> Result<(), Error> {
self.dispatcher.flush_all();
loop {
let elem = self.read_element()?;
self.dispatcher.dispatch(ReceiveElement(elem));
self.dispatcher.flush_all();
}
}
fn reset_stream(&self) {
self.transport.lock().unwrap().reset_stream()
}
fn read_element(&self) -> Result<Element, Error> {
self.transport.lock().unwrap().read_element()
}
fn write_element(&self, elem: &Element) -> Result<(), Error> {
self.transport.lock().unwrap().write_element(elem)
}
fn read_event(&self) -> Result<xml::reader::XmlEvent, Error> {
self.transport.lock().unwrap().read_event()
}
fn connect(&mut self, mut credentials: SaslCredentials) -> Result<(), Error> {
let features = self.wait_for_features()?;
let ms = &features.sasl_mechanisms.ok_or(Error::SaslError(Some("no SASL mechanisms".to_owned())))?;
fn wrap_err(err: String) -> Error { Error::SaslError(Some(err)) }
// TODO: better way for selecting these, enabling anonymous auth
let mut mechanism: Box<SaslMechanism> = if ms.contains("SCRAM-SHA-256-PLUS") && credentials.channel_binding != ChannelBinding::None {
Box::new(Scram::<Sha256>::from_credentials(credentials).map_err(wrap_err)?)
}
else if ms.contains("SCRAM-SHA-1-PLUS") && credentials.channel_binding != ChannelBinding::None {
Box::new(Scram::<Sha1>::from_credentials(credentials).map_err(wrap_err)?)
}
else if ms.contains("SCRAM-SHA-256") {
if credentials.channel_binding != ChannelBinding::None {
credentials.channel_binding = ChannelBinding::Unsupported;
}
Box::new(Scram::<Sha256>::from_credentials(credentials).map_err(wrap_err)?)
}
else if ms.contains("SCRAM-SHA-1") {
if credentials.channel_binding != ChannelBinding::None {
credentials.channel_binding = ChannelBinding::Unsupported;
}
Box::new(Scram::<Sha1>::from_credentials(credentials).map_err(wrap_err)?)
}
else if ms.contains("PLAIN") {
Box::new(Plain::from_credentials(credentials).map_err(wrap_err)?)
}
else {
return Err(Error::SaslError(Some("can't find a SASL mechanism to use".to_owned())));
};
let auth = mechanism.initial().map_err(|x| Error::SaslError(Some(x)))?;
let mut elem = Element::builder("auth")
.ns(ns::SASL)
.attr("mechanism", mechanism.name())
.build();
if !auth.is_empty() {
elem.append_text_node(base64::encode(&auth));
}
self.write_element(&elem)?;
loop {
let n = self.read_element()?;
if n.is("challenge", ns::SASL) {
let text = n.text();
let challenge = if text == "" {
Vec::new()
}
else {
base64::decode(&text)?
};
let response = mechanism.response(&challenge).map_err(|x| Error::SaslError(Some(x)))?;
let mut elem = Element::builder("response")
.ns(ns::SASL)
.build();
if !response.is_empty() {
elem.append_text_node(base64::encode(&response));
}
self.write_element(&elem)?;
}
else if n.is("success", ns::SASL) {
let text = n.text();
let data = if text == "" {
Vec::new()
}
else {
base64::decode(&text)?
};
mechanism.success(&data).map_err(|x| Error::SaslError(Some(x)))?;
self.reset_stream();
{
let mut g = self.transport.lock().unwrap();
C2S::init(&mut *g, &self.jid.domain, "after_sasl")?;
}
self.wait_for_features()?;
return Ok(());
}
else if n.is("failure", ns::SASL) {
let inner = SaslError::from_element(&n).map_err(|_| Error::SaslError(None))?;
return Err(Error::XmppSaslError(inner));
}
}
}
fn bind(&mut self) -> Result<(), Error> {
let mut elem = Element::builder("iq")
.attr("id", "bind")
.attr("type", "set")
.build();
let mut bind = Element::builder("bind")
.ns(ns::BIND)
.build();
if let Some(ref resource) = self.jid.resource {
let res = Element::builder("resource")
.ns(ns::BIND)
.append(resource.to_owned())
.build();
bind.append_child(res);
}
elem.append_child(bind);
self.write_element(&elem)?;
loop {
let n = self.read_element()?;
if n.is("iq", ns::CLIENT) && n.has_child("bind", ns::BIND) {
return Ok(());
}
}
}
fn wait_for_features(&mut self) -> Result<StreamFeatures, Error> {
// TODO: this is very ugly
loop {
let e = self.read_event()?;
match e {
ReaderEvent::StartElement { .. } => {
break;
},
_ => (),
}
}
loop {
let n = self.read_element()?;
if n.is("features", ns::STREAM) {
let mut features = StreamFeatures {
sasl_mechanisms: None,
};
if let Some(ms) = n.get_child("mechanisms", ns::SASL) {
let mut res = HashSet::new();
for cld in ms.children() {
res.insert(cld.text());
}
features.sasl_mechanisms = Some(res);
}
return Ok(features);
}
}
}
}

173
src/component.rs
View file

@ -1,173 +0,0 @@
use xml;
use jid::Jid;
use transport::{Transport, PlainTransport};
use error::Error;
use ns;
use plugin::{Plugin, PluginInit, PluginProxyBinding, PluginContainer, PluginRef};
use event::{Dispatcher, ReceiveElement, SendElement, Propagation, Priority, Event};
use connection::{Connection, Component2S};
use sha_1::{Sha1, Digest};
use minidom::Element;
use xml::reader::XmlEvent as ReaderEvent;
use std::fmt::Write;
use std::sync::{Mutex, Arc};
/// A builder for `Component`s.
pub struct ComponentBuilder {
jid: Jid,
secret: String,
host: Option<String>,
port: u16,
}
impl ComponentBuilder {
/// Creates a new builder for an XMPP component that will connect to `jid` with default parameters.
pub fn new(jid: Jid) -> ComponentBuilder {
ComponentBuilder {
jid: jid,
secret: "".to_owned(),
host: None,
port: 5347,
}
}
/// Sets the host to connect to.
pub fn host(mut self, host: String) -> ComponentBuilder {
self.host = Some(host);
self
}
/// Sets the port to connect to.
pub fn port(mut self, port: u16) -> ComponentBuilder {
self.port = port;
self
}
/// Sets the password to use.
pub fn password<P: Into<String>>(mut self, password: P) -> ComponentBuilder {
self.secret = password.into();
self
}
/// Connects to the server and returns a `Component` when succesful.
pub fn connect(self) -> Result<Component, Error> {
let host = &self.host.unwrap_or(self.jid.domain.clone());
let mut transport = PlainTransport::connect(host, self.port)?;
Component2S::init(&mut transport, &self.jid.domain, "stream_opening")?;
let dispatcher = Arc::new(Dispatcher::new());
let transport = Arc::new(Mutex::new(transport));
let plugin_container = Arc::new(PluginContainer::new());
let mut component = Component {
jid: self.jid.clone(),
transport: transport.clone(),
binding: PluginProxyBinding::new(dispatcher.clone(), plugin_container.clone(), self.jid),
plugin_container: plugin_container,
dispatcher: dispatcher,
};
component.dispatcher.register(Priority::Default, move |evt: &SendElement| {
let mut t = transport.lock().unwrap();
t.write_element(&evt.0).unwrap();
Propagation::Continue
});
component.connect(self.secret)?;
Ok(component)
}
}
/// An XMPP component.
pub struct Component {
jid: Jid,
transport: Arc<Mutex<PlainTransport>>,
plugin_container: Arc<PluginContainer>,
binding: PluginProxyBinding,
dispatcher: Arc<Dispatcher>,
}
impl Component {
/// Returns a reference to the `Jid` associated with this `Component`.
pub fn jid(&self) -> &Jid {
&self.jid
}
/// Registers a plugin.
pub fn register_plugin<P: Plugin + PluginInit + 'static>(&mut self, mut plugin: P) {
let binding = self.binding.clone();
plugin.bind(binding);
let p = Arc::new(plugin);
P::init(&self.dispatcher, p.clone());
self.plugin_container.register(p);
}
pub fn register_handler<E, F>(&mut self, pri: Priority, func: F)
where
E: Event,
F: Fn(&E) -> Propagation + 'static {
self.dispatcher.register(pri, func);
}
/// Returns the plugin given by the type parameter, if it exists, else panics.
pub fn plugin<P: Plugin>(&self) -> PluginRef<P> {
self.plugin_container.get::<P>().unwrap()
}
/// Returns the next event and flush the send queue.
pub fn main(&mut self) -> Result<(), Error> {
self.dispatcher.flush_all();
loop {
let elem = self.read_element()?;
self.dispatcher.dispatch(ReceiveElement(elem));
self.dispatcher.flush_all();
}
}
fn read_element(&self) -> Result<Element, Error> {
self.transport.lock().unwrap().read_element()
}
fn write_element(&self, elem: &Element) -> Result<(), Error> {
self.transport.lock().unwrap().write_element(elem)
}
fn read_event(&self) -> Result<xml::reader::XmlEvent, Error> {
self.transport.lock().unwrap().read_event()
}
fn connect(&mut self, secret: String) -> Result<(), Error> {
let mut sid = String::new();
loop {
let e = self.read_event()?;
match e {
ReaderEvent::StartElement { attributes, .. } => {
for attribute in attributes {
if attribute.name.namespace == None && attribute.name.local_name == "id" {
sid = attribute.value;
}
}
break;
},
_ => (),
}
}
let concatenated = format!("{}{}", sid, secret);
let mut hasher = Sha1::default();
hasher.input(concatenated.as_bytes());
let mut handshake = String::new();
for byte in hasher.result() {
write!(handshake, "{:02x}", byte)?;
}
let mut elem = Element::builder("handshake")
.ns(ns::COMPONENT_ACCEPT)
.build();
elem.append_text_node(handshake);
self.write_element(&elem)?;
loop {
let n = self.read_element()?;
if n.is("handshake", ns::COMPONENT_ACCEPT) {
return Ok(());
}
}
}
}

1
src/components/mod.rs
View file

@ -1 +0,0 @@
pub mod sasl_error;

84
src/components/sasl_error.rs
View file

@ -1,84 +0,0 @@
use ns;
use minidom::Element;
use util::FromElement;
#[derive(Clone, Debug)]
pub enum Condition {
Aborted,
AccountDisabled,
CredentialsExpired,
EncryptionRequired,
IncorrectEncoding,
InvalidAuthzid,
InvalidMechanism,
MalformedRequest,
MechanismTooWeak,
NotAuthorized,
TemporaryAuthFailure,
Unknown,
}
#[derive(Clone, Debug)]
pub struct SaslError {
condition: Condition,
text: Option<String>,
}
impl FromElement for SaslError {
type Err = ();
fn from_element(element: &Element) -> Result<SaslError, ()> {
if !element.is("failure", ns::SASL) {
return Err(());
}
let mut err = SaslError {
condition: Condition::Unknown,
text: None,
};
if let Some(text) = element.get_child("text", ns::SASL) {
let desc = text.text();
err.text = Some(desc);
}
if element.has_child("aborted", ns::SASL) {
err.condition = Condition::Aborted;
}
else if element.has_child("account-disabled", ns::SASL) {
err.condition = Condition::AccountDisabled;
}
else if element.has_child("credentials-expired", ns::SASL) {
err.condition = Condition::CredentialsExpired;
}
else if element.has_child("encryption-required", ns::SASL) {
err.condition = Condition::EncryptionRequired;
}
else if element.has_child("incorrect-encoding", ns::SASL) {
err.condition = Condition::IncorrectEncoding;
}
else if element.has_child("invalid-authzid", ns::SASL) {
err.condition = Condition::InvalidAuthzid;
}
else if element.has_child("malformed-request", ns::SASL) {
err.condition = Condition::MalformedRequest;
}
else if element.has_child("mechanism-too-weak", ns::SASL) {
err.condition = Condition::MechanismTooWeak;
}
else if element.has_child("not-authorized", ns::SASL) {
err.condition = Condition::NotAuthorized;
}
else if element.has_child("temporary-auth-failure", ns::SASL) {
err.condition = Condition::TemporaryAuthFailure;
}
else {
/* RFC 6120 section 6.5:
*
* However, because additional error conditions might be defined in
* the future, if an entity receives a SASL error condition that it
* does not understand then it MUST treat the unknown condition as
* a generic authentication failure, i.e., as equivalent to
* <not-authorized/> (Section 6.5.10). */
err.condition = Condition::NotAuthorized;
}
Ok(err)
}
}

61
src/connection.rs
View file

@ -1,61 +0,0 @@
use transport::Transport;
use error::Error;
use ns;
use xml::writer::XmlEvent as WriterEvent;
pub trait Connection {
type InitError;
type CloseError;
fn namespace() -> &'static str;
fn init<T: Transport>(transport: &mut T, domain: &str, id: &str) -> Result<(), Self::InitError>;
fn close<T: Transport>(transport: &mut T) -> Result<(), Self::CloseError>;
}
pub struct C2S;
impl Connection for C2S {
type InitError = Error;
type CloseError = Error;
fn namespace() -> &'static str { ns::CLIENT }
fn init<T: Transport>(transport: &mut T, domain: &str, id: &str) -> Result<(), Error> {
transport.write_event(WriterEvent::start_element("stream:stream")
.attr("to", domain)
.attr("id", id)
.default_ns(ns::CLIENT)
.ns("stream", ns::STREAM))?;
Ok(())
}
fn close<T: Transport>(transport: &mut T) -> Result<(), Error> {
transport.write_event(WriterEvent::end_element())?;
Ok(())
}
}
pub struct Component2S;
impl Connection for Component2S {
type InitError = Error;
type CloseError = Error;
fn namespace() -> &'static str { ns::COMPONENT_ACCEPT }
fn init<T: Transport>(transport: &mut T, domain: &str, id: &str) -> Result<(), Error> {
transport.write_event(WriterEvent::start_element("stream:stream")
.attr("to", domain)
.attr("id", id)
.default_ns(ns::COMPONENT_ACCEPT)
.ns("stream", ns::STREAM))?;
Ok(())
}
fn close<T: Transport>(transport: &mut T) -> Result<(), Error> {
transport.write_event(WriterEvent::end_element())?;
Ok(())
}
}

84
src/error.rs
View file

@ -1,84 +0,0 @@
//! Provides an `Error` for use in this crate.
use std::fmt::Error as FormatError;
use std::io;
use std::net::TcpStream;
use openssl::ssl::HandshakeError;
use openssl::error::ErrorStack;
use xml::reader::Error as XmlError;
use xml::writer::Error as EmitterError;
use minidom::Error as MinidomError;
use base64::DecodeError;
use components::sasl_error::SaslError;
/// An error which wraps a bunch of errors from different crates and the stdlib.
#[derive(Debug)]
pub enum Error {
XmlError(XmlError),
EmitterError(EmitterError),
IoError(io::Error),
HandshakeError(HandshakeError<TcpStream>),
OpenSslErrorStack(ErrorStack),
MinidomError(MinidomError),
Base64Error(DecodeError),
SaslError(Option<String>),
XmppSaslError(SaslError),
FormatError(FormatError),
StreamError,
EndOfDocument,
}
impl From<XmlError> for Error {
fn from(err: XmlError) -> Error {
Error::XmlError(err)
}
}
impl From<EmitterError> for Error {
fn from(err: EmitterError) -> Error {
Error::EmitterError(err)
}
}
impl From<io::Error> for Error {
fn from(err: io::Error) -> Error {
Error::IoError(err)
}
}
impl From<HandshakeError<TcpStream>> for Error {
fn from(err: HandshakeError<TcpStream>) -> Error {
Error::HandshakeError(err)
}
}
impl From<ErrorStack> for Error {
fn from(err: ErrorStack) -> Error {
Error::OpenSslErrorStack(err)
}
}
impl From<MinidomError> for Error {
fn from(err: MinidomError) -> Error {
Error::MinidomError(err)
}
}
impl From<DecodeError> for Error {
fn from(err: DecodeError) -> Error {
Error::Base64Error(err)
}
}
impl From<FormatError> for Error {
fn from(err: FormatError) -> Error {
Error::FormatError(err)
}
}

229
src/event.rs
View file

@ -1,229 +0,0 @@
use std::marker::PhantomData;
use std::any::{TypeId, Any};
use std::fmt::Debug;
use std::collections::BTreeMap;
use std::cmp::Ordering;
use std::mem;
use std::sync::Mutex;
use minidom::Element;
/// A marker trait which marks all events.
pub trait Event: Any + Debug {}
/// A trait which is implemented for all event handlers.
trait EventHandler: Any {
/// Handle an event, returns whether to propagate the event to the remaining handlers.
fn handle(&self, event: &AbstractEvent) -> Propagation;
}
/// An abstract event.
pub struct AbstractEvent {
inner: Box<Any>,
}
impl AbstractEvent {
/// Creates an abstract event from a concrete event.
pub fn new<E: Event>(event: E) -> AbstractEvent {
AbstractEvent {
inner: Box::new(event),
}
}
/// Downcasts this abstract event into a concrete event.
pub fn downcast<E: Event + 'static>(&self) -> Option<&E> {
self.inner.downcast_ref::<E>()
}
/// Checks whether this abstract event is a specific concrete event.
pub fn is<E: Event + 'static>(&self) -> bool {
self.inner.is::<E>()
}
}
struct Record<P, T>(P, T);
impl<P: PartialEq, T> PartialEq for Record<P, T> {
fn eq(&self, other: &Record<P, T>) -> bool {
self.0 == other.0
}
}
impl<P: Eq, T> Eq for Record<P, T> {}
impl<P: PartialOrd, T> PartialOrd for Record<P, T> {
fn partial_cmp(&self, other: &Record<P, T>) -> Option<Ordering> {
self.0.partial_cmp(&other.0)
}
}
impl<P: Ord, T> Ord for Record<P, T> {
fn cmp(&self, other: &Record<P, T>) -> Ordering {
self.0.cmp(&other.0)
}
}
/// An enum representing whether to keep propagating an event or to stop the propagation.
pub enum Propagation {
/// Stop the propagation of the event, the remaining handlers will not get invoked.
Stop,
/// Continue propagating the event.
Continue,
}
/// An event dispatcher, this takes care of dispatching events to their respective handlers.
pub struct Dispatcher {
handlers: Mutex<BTreeMap<TypeId, Vec<Record<Priority, Box<EventHandler>>>>>,
queue: Mutex<Vec<(TypeId, AbstractEvent)>>,
}
impl Dispatcher {
/// Create a new `Dispatcher`.
pub fn new() -> Dispatcher {
Dispatcher {
handlers: Mutex::new(BTreeMap::new()),
queue: Mutex::new(Vec::new()),
}
}
/// Register an event handler.
pub fn register<E, F>(&self, priority: Priority, func: F)
where
E: Event,
F: Fn(&E) -> Propagation + 'static {
struct Handler<E, F> where E: Event, F: Fn(&E) -> Propagation {
func: F,
_marker: PhantomData<E>,
}
impl<E: Event, F: Fn(&E) -> Propagation + 'static> EventHandler for Handler<E, F> {
fn handle(&self, evt: &AbstractEvent) -> Propagation {
if let Some(e) = evt.downcast::<E>() {
(self.func)(e)
}
else {
Propagation::Continue
}
}
}
let handler: Box<EventHandler> = Box::new(Handler {
func: func,
_marker: PhantomData,
}) as Box<EventHandler>;
let mut guard = self.handlers.lock().unwrap();
let ent = guard.entry(TypeId::of::<E>())
.or_insert_with(|| Vec::new());
ent.push(Record(priority, handler));
ent.sort();
}
/// Append an event to the queue.
pub fn dispatch<E>(&self, event: E) where E: Event {
self.queue.lock().unwrap().push((TypeId::of::<E>(), AbstractEvent::new(event)));
}
/// Flush all events in the queue so they can be handled by their respective handlers.
/// Returns whether there are still pending events.
pub fn flush(&self) -> bool {
let mut q = Vec::new();
{
let mut my_q = self.queue.lock().unwrap();
mem::swap(my_q.as_mut(), &mut q);
}
'evts: for (t, evt) in q {
if let Some(handlers) = self.handlers.lock().unwrap().get_mut(&t) {
for &mut Record(_, ref mut handler) in handlers {
match handler.handle(&evt) {
Propagation::Stop => { continue 'evts; },
Propagation::Continue => (),
}
}
}
}
!self.queue.lock().unwrap().is_empty()
}
/// Flushes all events, like `flush`, but keeps doing this until there is nothing left in the
/// queue.
pub fn flush_all(&self) {
while self.flush() {}
}
}
#[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
pub enum Priority {
Max,
Default,
Min,
}
impl Default for Priority {
fn default() -> Priority {
Priority::Default
}
}
#[derive(Debug)]
pub struct SendElement(pub Element);
impl Event for SendElement {}
#[derive(Debug)]
pub struct ReceiveElement(pub Element);
impl Event for ReceiveElement {}
#[cfg(test)]
mod tests {
use super::*;
#[test]
#[should_panic(expected = "success")]
fn test() {
let disp = Dispatcher::new();
#[derive(Debug)]
struct MyEvent {
should_be_42: u32,
}
impl Event for MyEvent {}
disp.register(Priority::Max, |evt: &MyEvent| {
if evt.should_be_42 == 42 {
Propagation::Continue
}
else {
Propagation::Stop
}
});
disp.register(Priority::Min, |_: &MyEvent| {
panic!("should not be called");
});
disp.register(Priority::Default, |evt: &MyEvent| {
if evt.should_be_42 == 42 {
panic!("success");
}
else {
panic!("not 42");
}
});
disp.register(Priority::Min, |_: &MyEvent| {
panic!("should not be called");
});
disp.dispatch(MyEvent {
should_be_42: 39,
});
disp.dispatch(MyEvent {
should_be_42: 42,
});
disp.flush();
}
}

403
src/lib.rs
View file

@ -1,25 +1,382 @@
extern crate xml;
extern crate xmpp_parsers;
extern crate openssl;
extern crate minidom;
extern crate base64;
extern crate sha_1;
extern crate chrono;
extern crate try_from;
pub extern crate jid;
pub extern crate sasl;
// Copyright (c) 2019 Emmanuel Gil Peyrot <linkmauve@linkmauve.fr>
//
// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at http://mozilla.org/MPL/2.0/.
pub mod ns;
pub mod transport;
pub mod error;
pub mod client;
pub mod component;
pub mod plugin;
#[macro_use] pub mod plugin_macro;
pub mod event;
pub mod plugins;
pub mod connection;
pub mod util;
pub mod components;
#![deny(bare_trait_objects)]
mod locked_io;
use std::str::FromStr;
use std::rc::Rc;
use std::cell::RefCell;
use std::convert::TryFrom;
use futures::{Future,Stream, Sink, sync::mpsc};
use tokio_xmpp::{
Client as TokioXmppClient,
Event as TokioXmppEvent,
Packet,
};
use xmpp_parsers::{
bookmarks::{
Autojoin,
Conference as ConferenceBookmark,
Storage as Bookmarks,
},
caps::{compute_disco, hash_caps, Caps},
disco::{DiscoInfoQuery, DiscoInfoResult, Feature, Identity},
hashes::Algo,
iq::{Iq, IqType},
message::{Message, MessageType, Body},
muc::{
Muc,
user::{MucUser, Status},
},
ns,
presence::{Presence, Type as PresenceType},
pubsub::{
event::PubSubEvent,
pubsub::PubSub,
},
roster::{Roster, Item as RosterItem},
stanza_error::{StanzaError, ErrorType, DefinedCondition},
Jid, BareJid, FullJid, JidParseError,
};
mod avatar;
pub type Error = tokio_xmpp::Error;
#[derive(Debug)]
pub enum ClientType {
Bot,
Pc,
}
impl Default for ClientType {
fn default() -> Self {
ClientType::Bot
}
}
impl ToString for ClientType {
fn to_string(&self) -> String {
String::from(
match self {
ClientType::Bot => "bot",
ClientType::Pc => "pc",
}
)
}
}
#[derive(PartialEq)]
pub enum ClientFeature {
Avatars,
ContactList,
JoinRooms,
}
#[derive(Debug)]
pub enum Event {
Online,
Disconnected,
ContactAdded(RosterItem),
ContactRemoved(RosterItem),
ContactChanged(RosterItem),
AvatarRetrieved(Jid, String),
OpenRoomBookmark(ConferenceBookmark),
RoomJoined(BareJid),
RoomLeft(BareJid),
}
#[derive(Default)]
pub struct ClientBuilder<'a> {
jid: &'a str,
password: &'a str,
website: String,
default_nick: String,
disco: (ClientType, String),
features: Vec<ClientFeature>,
}
impl ClientBuilder<'_> {
pub fn new<'a>(jid: &'a str, password: &'a str) -> ClientBuilder<'a> {
ClientBuilder {
jid,
password,
website: String::from("https://gitlab.com/xmpp-rs/tokio-xmpp"),
default_nick: String::from("xmpp-rs"),
disco: (ClientType::default(), String::from("tokio-xmpp")),
features: vec![],
}
}
pub fn set_client(mut self, type_: ClientType, name: &str) -> Self {
self.disco = (type_, String::from(name));
self
}
pub fn set_website(mut self, url: &str) -> Self {
self.website = String::from(url);
self
}
pub fn set_default_nick(mut self, nick: &str) -> Self {
self.default_nick = String::from(nick);
self
}
pub fn enable_feature(mut self, feature: ClientFeature) -> Self {
self.features.push(feature);
self
}
fn make_disco(&self) -> DiscoInfoResult {
let identities = vec![Identity::new("client", self.disco.0.to_string(),
"en", self.disco.1.to_string())];
let mut features = vec![
Feature::new(ns::DISCO_INFO),
];
if self.features.contains(&ClientFeature::Avatars) {
features.push(Feature::new(format!("{}+notify", ns::AVATAR_METADATA)));
}
if self.features.contains(&ClientFeature::JoinRooms) {
features.push(Feature::new(format!("{}+notify", ns::BOOKMARKS)));
}
DiscoInfoResult {
node: None,
identities,
features,
extensions: vec![],
}
}
fn make_initial_presence(disco: &DiscoInfoResult, node: &str) -> Presence {
let caps_data = compute_disco(disco);
let hash = hash_caps(&caps_data, Algo::Sha_1).unwrap();
let caps = Caps::new(node, hash);
let mut presence = Presence::new(PresenceType::None);
presence.add_payload(caps);
presence
}
pub fn build(
self,
) -> Result<(Agent, impl Stream<Item = Event, Error = tokio_xmpp::Error>), JidParseError> {
let client = TokioXmppClient::new(self.jid, self.password)?;
Ok(self.build_impl(client))
}
// This function is meant to be used for testing build
pub(crate) fn build_impl<S>(
self,
stream: S,
) -> (Agent, impl Stream<Item = Event, Error = tokio_xmpp::Error>)
where
S: Stream<Item = tokio_xmpp::Event, Error = tokio_xmpp::Error>
+ Sink<SinkItem = tokio_xmpp::Packet, SinkError = tokio_xmpp::Error>,
{
let disco = self.make_disco();
let node = self.website;
let (sender_tx, sender_rx) = mpsc::unbounded();
let client = stream;
let (sink, stream) = client.split();
let reader = {
let mut sender_tx = sender_tx.clone();
let jid = self.jid.to_owned();
stream.map(move |event| {
// Helper function to send an iq error.
let mut events = Vec::new();
let send_error = |to, id, type_, condition, text: &str| {
let error = StanzaError::new(type_, condition, "en", text);
let iq = Iq::from_error(id, error)
.with_to(to)
.into();
sender_tx.unbounded_send(Packet::Stanza(iq)).unwrap();
};
match event {
TokioXmppEvent::Online => {
let presence = ClientBuilder::make_initial_presence(&disco, &node).into();
let packet = Packet::Stanza(presence);
sender_tx.unbounded_send(packet)
.unwrap();
events.push(Event::Online);
// TODO: only send this when the ContactList feature is enabled.
let iq = Iq::from_get("roster", Roster { ver: None, items: vec![] })
.into();
sender_tx.unbounded_send(Packet::Stanza(iq)).unwrap();
}
TokioXmppEvent::Disconnected => {
events.push(Event::Disconnected);
}
TokioXmppEvent::Stanza(stanza) => {
if stanza.is("iq", "jabber:client") {
let iq = Iq::try_from(stanza).unwrap();
if let IqType::Get(payload) = iq.payload {
if payload.is("query", ns::DISCO_INFO) {
let query = DiscoInfoQuery::try_from(payload);
match query {
Ok(query) => {
let mut disco_info = disco.clone();
disco_info.node = query.node;
let iq = Iq::from_result(iq.id, Some(disco_info))
.with_to(iq.from.unwrap())
.into();
sender_tx.unbounded_send(Packet::Stanza(iq)).unwrap();
},
Err(err) => {
send_error(iq.from.unwrap(), iq.id, ErrorType::Modify, DefinedCondition::BadRequest, &format!("{}", err));
},
}
} else {
// We MUST answer unhandled get iqs with a service-unavailable error.
send_error(iq.from.unwrap(), iq.id, ErrorType::Cancel, DefinedCondition::ServiceUnavailable, "No handler defined for this kind of iq.");
}
} else if let IqType::Result(Some(payload)) = iq.payload {
// TODO: move private iqs like this one somewhere else, for
// security reasons.
if payload.is("query", ns::ROSTER) && iq.from.is_none() {
let roster = Roster::try_from(payload).unwrap();
for item in roster.items.into_iter() {
events.push(Event::ContactAdded(item));
}
} else if payload.is("pubsub", ns::PUBSUB) {
let pubsub = PubSub::try_from(payload).unwrap();
let from =
iq.from.clone().unwrap_or_else(|| Jid::from_str(&jid).unwrap());
if let PubSub::Items(items) = pubsub {
if items.node.0 == ns::AVATAR_DATA {
let new_events = avatar::handle_data_pubsub_iq(&from, &items);
events.extend(new_events);
}
}
}
} else if let IqType::Set(_) = iq.payload {
// We MUST answer unhandled set iqs with a service-unavailable error.
send_error(iq.from.unwrap(), iq.id, ErrorType::Cancel, DefinedCondition::ServiceUnavailable, "No handler defined for this kind of iq.");
}
} else if stanza.is("message", "jabber:client") {
let message = Message::try_from(stanza).unwrap();
let from = message.from.clone().unwrap();
for child in message.payloads {
if child.is("event", ns::PUBSUB_EVENT) {
let event = PubSubEvent::try_from(child).unwrap();
if let PubSubEvent::PublishedItems { node, items } = event {
if node.0 == ns::AVATAR_METADATA {
let new_events = avatar::handle_metadata_pubsub_event(&from, &mut sender_tx, items);
events.extend(new_events);
} else if node.0 == ns::BOOKMARKS {
// TODO: Check that our bare JID is the sender.
assert_eq!(items.len(), 1);
let item = items.clone().pop().unwrap();
let payload = item.payload.clone().unwrap();
let bookmarks = match Bookmarks::try_from(payload) {
Ok(bookmarks) => bookmarks,
// XXX: Don’t panic…
Err(err) => panic!("… {}", err),
};
for bookmark in bookmarks.conferences {
if bookmark.autojoin == Autojoin::True {
events.push(Event::OpenRoomBookmark(bookmark));
}
}
}
}
}
}
} else if stanza.is("presence", "jabber:client") {
let presence = Presence::try_from(stanza).unwrap();
let from: BareJid = match presence.from.clone().unwrap() {
Jid::Full(FullJid { node, domain, .. }) => BareJid { node, domain },
Jid::Bare(bare) => bare,
};
for payload in presence.payloads.into_iter() {
let muc_user = match MucUser::try_from(payload) {
Ok(muc_user) => muc_user,
_ => continue
};
for status in muc_user.status.into_iter() {
if status == Status::SelfPresence {
events.push(Event::RoomJoined(from.clone()));
break;
}
}
}
} else if stanza.is("error", "http://etherx.jabber.org/streams") {
println!("Received a fatal stream error: {}", String::from(&stanza));
} else {
panic!("Unknown stanza: {}", String::from(&stanza));
}
}
}
futures::stream::iter_ok(events)
})
.flatten()
};
let sender = sender_rx
.map_err(|e| panic!("Sink error: {:?}", e))
.forward(sink)
.map(|(rx, mut sink)| {
drop(rx);
let _ = sink.close();
None
});
// TODO is this correct?
// Some(Error) means a real error
// None means the end of the sender stream and can be ignored
let future = reader
.map(Some)
.select(sender.into_stream())
.filter_map(|x| x);
let agent = Agent {
sender_tx,
default_nick: Rc::new(RefCell::new(self.default_nick)),
};
(agent, future)
}
}
#[derive(Clone)]
pub struct Agent {
sender_tx: mpsc::UnboundedSender<Packet>,
default_nick: Rc<RefCell<String>>,
}
impl Agent {
pub fn join_room(&mut self, room: BareJid, nick: Option<String>, password: Option<String>,
lang: &str, status: &str) {
let mut muc = Muc::new();
if let Some(password) = password {
muc = muc.with_password(password);
}
let nick = nick.unwrap_or_else(|| self.default_nick.borrow().clone());
let room_jid = room.with_resource(nick);
let mut presence = Presence::new(PresenceType::None)
.with_to(Some(Jid::Full(room_jid)));
presence.add_payload(muc);
presence.set_status(String::from(lang), String::from(status));
let presence = presence.into();
self.sender_tx.unbounded_send(Packet::Stanza(presence))
.unwrap();
}
pub fn send_message(&mut self, recipient: Jid, type_: MessageType, lang: &str, text: &str) {
let mut message = Message::new(Some(recipient));
message.type_ = type_;
message.bodies.insert(String::from(lang), Body(String::from(text)));
let message = message.into();
self.sender_tx.unbounded_send(Packet::Stanza(message))
.unwrap();
}
}
>>>>>>> lm-master

37
src/locked_io.rs
View file

@ -1,37 +0,0 @@
use std::io;
use std::io::prelude::*;
use std::sync::{Arc, Mutex};
pub struct LockedIO<T>(Arc<Mutex<T>>);
impl<T> LockedIO<T> {
pub fn from(inner: Arc<Mutex<T>>) -> LockedIO<T> {
LockedIO(inner)
}
}
impl<T> Clone for LockedIO<T> {
fn clone(&self) -> LockedIO<T> {
LockedIO(self.0.clone())
}
}
impl<T: Write> io::Write for LockedIO<T> {
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
let mut inner = self.0.lock().unwrap(); // TODO: make safer
inner.write(buf)
}
fn flush(&mut self) -> io::Result<()> {
let mut inner = self.0.lock().unwrap(); // TODO: make safer
inner.flush()
}
}
impl<T: Read> io::Read for LockedIO<T> {
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
let mut inner = self.0.lock().unwrap(); // TODO: make safer
inner.read(buf)
}
}

10
src/ns.rs
View file

@ -1,10 +0,0 @@
//! Provides constants for namespaces.
pub const CLIENT: &'static str = "jabber:client";
pub const COMPONENT_ACCEPT: &'static str = "jabber:component:accept";
pub const STREAM: &'static str = "http://etherx.jabber.org/streams";
pub const TLS: &'static str = "urn:ietf:params:xml:ns:xmpp-tls";
pub const SASL: &'static str = "urn:ietf:params:xml:ns:xmpp-sasl";
pub const BIND: &'static str = "urn:ietf:params:xml:ns:xmpp-bind";
pub const STANZAS: &'static str = "urn:ietf:params:xml:ns:xmpp-stanzas";
pub const PING: &'static str = "urn:xmpp:ping";

189
src/plugin.rs
View file

@ -1,189 +0,0 @@
//! Provides the plugin infrastructure.
use event::{Event, Dispatcher, SendElement, Priority, Propagation};
use std::any::{Any, TypeId};
use std::collections::HashMap;
use std::sync::{RwLock, Arc};
use std::sync::atomic::{AtomicUsize, Ordering, ATOMIC_USIZE_INIT};
use std::marker::PhantomData;
use std::ops::Deref;
use std::convert::AsRef;
use std::mem;
use minidom::Element;
use jid::Jid;
pub struct PluginContainer {
plugins: RwLock<HashMap<TypeId, Arc<Plugin>>>,
}
impl PluginContainer {
pub fn new() -> PluginContainer {
PluginContainer {
plugins: RwLock::new(HashMap::new()),
}
}
pub fn register<P: Plugin + 'static>(&self, plugin: Arc<P>) {
let mut guard = self.plugins.write().unwrap();
if guard.insert(TypeId::of::<P>(), plugin as Arc<Plugin>).is_some() {
panic!("registering a plugin that's already registered");
}
}
pub fn get<P: Plugin>(&self) -> Option<PluginRef<P>> {
let guard = self.plugins.read().unwrap();
let arc = guard.get(&TypeId::of::<P>());
arc.map(|arc| PluginRef {
inner: arc.clone(),
_marker: PhantomData
})
}
}
#[derive(Clone)]
pub struct PluginRef<P: Plugin> {
inner: Arc<Plugin>,
_marker: PhantomData<P>,
}
impl<P: Plugin> Deref for PluginRef<P> {
type Target = P;
fn deref(&self) -> &P {
self.inner.as_any().downcast_ref::<P>().expect("plugin downcast failure")
}
}
impl<P: Plugin> AsRef<P> for PluginRef<P> {
fn as_ref(&self) -> &P {
self.inner.as_any().downcast_ref::<P>().expect("plugin downcast failure")
}
}
#[derive(Clone)]
pub struct PluginProxyBinding {
dispatcher: Arc<Dispatcher>,
plugin_container: Arc<PluginContainer>,
jid: Jid,
next_id: Arc<AtomicUsize>,
}
impl PluginProxyBinding {
pub fn new(dispatcher: Arc<Dispatcher>, plugin_container: Arc<PluginContainer>, jid: Jid) -> PluginProxyBinding {
PluginProxyBinding {
dispatcher: dispatcher,
plugin_container: plugin_container,
jid: jid,
next_id: Arc::new(ATOMIC_USIZE_INIT),
}
}
}
pub enum PluginProxy {
Unbound,
BoundTo(PluginProxyBinding),
}
impl PluginProxy {
/// Returns a new `PluginProxy`.
pub fn new() -> PluginProxy {
PluginProxy::Unbound
}
/// Binds the `PluginProxy` to a `PluginProxyBinding`.
pub fn bind(&mut self, inner: PluginProxyBinding) {
if let PluginProxy::BoundTo(_) = *self {
panic!("trying to bind an already bound plugin proxy!");
}
mem::replace(self, PluginProxy::BoundTo(inner));
}
fn with_binding<R, F: FnOnce(&PluginProxyBinding) -> R>(&self, f: F) -> R {
match *self {
PluginProxy::Unbound => {
panic!("trying to use an unbound plugin proxy!");
},
PluginProxy::BoundTo(ref binding) => {
f(binding)
},
}
}
/// Dispatches an event.
pub fn dispatch<E: Event>(&self, event: E) {
self.with_binding(move |binding| {
// TODO: proper error handling
binding.dispatcher.dispatch(event);
});
}
/// Registers an event handler.
pub fn register_handler<E, F>(&self, priority: Priority, func: F)
where
E: Event,
F: Fn(&E) -> Propagation + 'static {
self.with_binding(move |binding| {
// TODO: proper error handling
binding.dispatcher.register(priority, func);
});
}
/// Tries to get another plugin.
pub fn plugin<P: Plugin>(&self) -> Option<PluginRef<P>> {
self.with_binding(|binding| {
binding.plugin_container.get::<P>()
})
}
/// Sends a stanza.
pub fn send(&self, elem: Element) {
self.dispatch(SendElement(elem));
}
/// Get our own JID.
pub fn get_own_jid(&self) -> Jid {
self.with_binding(|binding| {
binding.jid.clone()
})
}
/// Get a new id.
pub fn gen_id(&self) -> String {
self.with_binding(|binding| {
format!("{}", binding.next_id.fetch_add(1, Ordering::SeqCst))
})
}
}
/// A trait whch all plugins should implement.
pub trait Plugin: Any + PluginAny {
/// Gets a mutable reference to the inner `PluginProxy`.
fn get_proxy(&mut self) -> &mut PluginProxy;
#[doc(hidden)]
fn bind(&mut self, inner: PluginProxyBinding) {
self.get_proxy().bind(inner);
}
}
pub trait PluginInit {
fn init(dispatcher: &Dispatcher, me: Arc<Plugin>);
}
pub trait PluginAny {
fn as_any(&self) -> &Any;
}
impl<T: Any + Sized + Plugin> PluginAny for T {
fn as_any(&self) -> &Any { self }
}

28
src/plugin_macro.rs
View file

@ -1,28 +0,0 @@
#[macro_export]
macro_rules! impl_plugin {
($plugin:ty, $proxy:ident, [$(($evt:ty, $pri:expr) => $func:ident),*]) => {
impl $crate::plugin::Plugin for $plugin {
fn get_proxy(&mut self) -> &mut $crate::plugin::PluginProxy {
&mut self.$proxy
}
}
#[allow(unused_variables)]
impl $crate::plugin::PluginInit for $plugin {
fn init( dispatcher: &$crate::event::Dispatcher
, me: ::std::sync::Arc<$crate::plugin::Plugin>) {
$(
let new_arc = me.clone();
dispatcher.register($pri, move |e: &$evt| {
let p = new_arc.as_any().downcast_ref::<$plugin>().unwrap();
p . $func(e)
});
)*
}
}
};
($plugin:ty, $proxy:ident, [$(($evt:ty, $pri:expr) => $func:ident),*,]) => {
impl_plugin!($plugin, $proxy, [$(($evt, $pri) => $func),*]);
};
}

137
src/plugins/caps.rs
View file

@ -1,137 +0,0 @@
use std::collections::HashMap;
use try_from::TryFrom;
use std::sync::{Mutex, Arc};
use plugin::PluginProxy;
use event::{Event, Priority, Propagation};
use jid::Jid;
use base64;
use plugins::stanza::{Presence, Iq};
use plugins::disco::DiscoInfoResult;
use xmpp_parsers::presence::Type as PresenceType;
use xmpp_parsers::iq::IqType;
use xmpp_parsers::disco::{DiscoInfoQuery, DiscoInfoResult as DiscoInfoResult_};
use xmpp_parsers::caps::Caps;
#[derive(Debug)]
pub struct DiscoInfoRequest {
pub from: Jid,
pub id: String,
pub node: Option<String>,
}
impl Event for DiscoInfoRequest {}
pub struct CapsPlugin {
proxy: PluginProxy,
pending: Arc<Mutex<HashMap<Jid, (String, String)>>>,
cache: Arc<Mutex<HashMap<(Jid, String), DiscoInfoResult_>>>,
}
impl CapsPlugin {
pub fn new() -> CapsPlugin {
CapsPlugin {
proxy: PluginProxy::new(),
pending: Arc::new(Mutex::new(HashMap::new())),
cache: Arc::new(Mutex::new(HashMap::new())),
}
}
fn handle_presence(&self, presence: &Presence) -> Propagation {
let presence = presence.clone();
match presence.type_ {
PresenceType::None => for payload in presence.payloads {
let caps = match Caps::try_from(payload) {
Ok(caps) => caps,
Err(_) => continue,
};
let recipient = presence.from.unwrap();
let node = format!("{}#{}", caps.node, base64::encode(&caps.hash.hash));
{
let cache = self.cache.lock().unwrap();
if cache.contains_key(&(recipient.clone(), node.clone())) {
break;
}
}
let id = self.proxy.gen_id();
{
let mut pending = self.pending.lock().unwrap();
pending.insert(recipient.clone(), (id.clone(), node.clone()));
}
let disco = DiscoInfoQuery {
node: Some(node),
};
self.proxy.send(Iq {
to: Some(recipient),
from: None,
id: Some(id),
payload: IqType::Get(disco.into()),
}.into());
break;
},
PresenceType::Unavailable
| PresenceType::Error => {
let recipient = presence.from.unwrap();
let mut pending = self.pending.lock().unwrap();
let previous = pending.remove(&recipient);
if previous.is_none() {
// This wasn’t one of our requests.
return Propagation::Continue;
}
// TODO: maybe add a negative cache?
},
_ => (),
}
Propagation::Continue
}
fn handle_result(&self, result: &DiscoInfoResult) -> Propagation {
let from = result.from.clone();
let mut pending = self.pending.lock().unwrap();
let previous = pending.remove(&from.clone());
if let Some((id, node)) = previous {
if id != result.id {
return Propagation::Continue;
}
if Some(node.clone()) != result.disco.node {
// TODO: make that a debug log.
println!("Wrong node in result!");
return Propagation::Continue;
}
{
let mut cache = self.cache.lock().unwrap();
cache.insert((from, node), result.disco.clone());
}
} else {
// TODO: make that a debug log.
println!("No such request from us.");
return Propagation::Continue;
}
Propagation::Stop
}
// This is only for errors.
// TODO: also do the same thing for timeouts.
fn handle_iq(&self, iq: &Iq) -> Propagation {
let iq = iq.clone();
if let IqType::Error(_) = iq.payload {
let from = iq.from.unwrap();
let mut pending = self.pending.lock().unwrap();
let previous = pending.remove(&from.clone());
if previous.is_none() {
// This wasn’t one of our requests.
return Propagation::Continue;
}
// TODO: maybe add a negative cache?
return Propagation::Stop;
}
Propagation::Continue
}
}
impl_plugin!(CapsPlugin, proxy, [
(Presence, Priority::Default) => handle_presence,
(Iq, Priority::Default) => handle_iq,
(DiscoInfoResult, Priority::Default) => handle_result,
]);

141
src/plugins/disco.rs
View file

@ -1,141 +0,0 @@
use try_from::TryFrom;
use std::sync::{Mutex, Arc};
use plugin::PluginProxy;
use event::{Event, Priority, Propagation};
use jid::Jid;
use plugins::stanza::Iq;
use xmpp_parsers::iq::IqType;
use xmpp_parsers::disco::{DiscoInfoQuery, DiscoInfoResult as DiscoInfoResult_, Identity, Feature};
use xmpp_parsers::data_forms::DataForm;
use xmpp_parsers::ns;
#[derive(Debug)]
pub struct DiscoInfoRequest {
pub from: Jid,
pub id: String,
pub node: Option<String>,
}
#[derive(Debug)]
pub struct DiscoInfoResult {
pub from: Jid,
pub id: String,
pub disco: DiscoInfoResult_,
}
impl Event for DiscoInfoRequest {}
impl Event for DiscoInfoResult {}
pub struct DiscoPlugin {
proxy: PluginProxy,
cached_disco: Arc<Mutex<DiscoInfoResult_>>,
}
impl DiscoPlugin {
pub fn new(category: &str, type_: &str, lang: &str, name: &str) -> DiscoPlugin {
DiscoPlugin {
proxy: PluginProxy::new(),
cached_disco: Arc::new(Mutex::new(DiscoInfoResult_ {
node: None,
identities: vec!(Identity {
category: category.to_owned(),
type_: type_.to_owned(),
lang: Some(lang.to_owned()),
name: Some(name.to_owned())
}),
features: vec!(Feature { var: String::from(ns::DISCO_INFO) }),
extensions: vec!(),
})),
}
}
pub fn add_identity(&self, category: &str, type_: &str, lang: Option<&str>, name: Option<&str>) {
let mut cached_disco = self.cached_disco.lock().unwrap();
cached_disco.identities.push(Identity {
category: category.to_owned(),
type_: type_.to_owned(),
lang: lang.and_then(|lang| Some(lang.to_owned())),
name: name.and_then(|name| Some(name.to_owned())),
});
}
pub fn remove_identity(&self, category: &str, type_: &str, lang: Option<&str>, name: Option<&str>) {
let mut cached_disco = self.cached_disco.lock().unwrap();
cached_disco.identities.retain(|identity| {
identity.category != category ||
identity.type_ != type_ ||
identity.lang != lang.and_then(|lang| Some(lang.to_owned())) ||
identity.name != name.and_then(|name| Some(name.to_owned()))
});
}
pub fn add_feature(&self, var: &str) {
let mut cached_disco = self.cached_disco.lock().unwrap();
cached_disco.features.push(Feature { var: String::from(var) });
}
pub fn remove_feature(&self, var: &str) {
let mut cached_disco = self.cached_disco.lock().unwrap();
cached_disco.features.retain(|feature| feature.var != var);
}
pub fn add_extension(&self, extension: DataForm) {
let mut cached_disco = self.cached_disco.lock().unwrap();
cached_disco.extensions.push(extension);
}
pub fn remove_extension(&self, form_type: &str) {
let mut cached_disco = self.cached_disco.lock().unwrap();
cached_disco.extensions.retain(|extension| {
extension.form_type != Some(form_type.to_owned())
});
}
fn handle_iq(&self, iq: &Iq) -> Propagation {
let iq = iq.clone();
if let IqType::Get(payload) = iq.payload {
if let Ok(disco) = DiscoInfoQuery::try_from(payload) {
self.proxy.dispatch(DiscoInfoRequest {
from: iq.from.unwrap(),
id: iq.id.unwrap(),
node: disco.node,
});
return Propagation::Stop;
}
} else if let IqType::Result(Some(payload)) = iq.payload {
if let Ok(disco) = DiscoInfoResult_::try_from(payload) {
self.proxy.dispatch(DiscoInfoResult {
from: iq.from.unwrap(),
id: iq.id.unwrap(),
disco: disco,
});
return Propagation::Stop;
}
}
Propagation::Continue
}
fn reply_disco_info(&self, request: &DiscoInfoRequest) -> Propagation {
let payload = if request.node.is_none() {
let cached_disco = self.cached_disco.lock().unwrap().clone();
IqType::Result(Some(cached_disco.into()))
} else {
// TODO: handle the requests on nodes too.
return Propagation::Continue;
};
self.proxy.send(Iq {
from: None,
to: Some(request.from.to_owned()),
id: Some(request.id.to_owned()),
payload,
}.into());
Propagation::Stop
}
}
impl_plugin!(DiscoPlugin, proxy, [
(Iq, Priority::Default) => handle_iq,
(DiscoInfoRequest, Priority::Default) => reply_disco_info,
]);

195
src/plugins/ibb.rs
View file

@ -1,195 +0,0 @@
use std::collections::{HashMap, BTreeMap};
use std::collections::hash_map::Entry;
use try_from::TryFrom;
use std::sync::{Mutex, Arc};
use plugin::PluginProxy;
use event::{Event, Priority, Propagation};
use jid::Jid;
use plugins::stanza::Iq;
use plugins::disco::DiscoPlugin;
use xmpp_parsers::iq::{IqType, IqSetPayload};
use xmpp_parsers::ibb::{IBB, Stanza};
use xmpp_parsers::stanza_error::{StanzaError, ErrorType, DefinedCondition};
use xmpp_parsers::ns;
#[derive(Debug, Clone)]
pub struct Session {
stanza: Stanza,
block_size: u16,
cur_seq: u16,
}
#[derive(Debug)]
pub struct IbbOpen {
pub session: Session,
}
#[derive(Debug)]
pub struct IbbData {
pub session: Session,
pub data: Vec<u8>,
}
#[derive(Debug)]
pub struct IbbClose {
pub session: Session,
}
impl Event for IbbOpen {}
impl Event for IbbData {}
impl Event for IbbClose {}
fn generate_error(type_: ErrorType, defined_condition: DefinedCondition, text: &str) -> StanzaError {
StanzaError {
type_: type_,
defined_condition: defined_condition,
texts: {
let mut texts = BTreeMap::new();
texts.insert(String::new(), String::from(text));
texts
},
by: None,
other: None,
}
}
pub struct IbbPlugin {
proxy: PluginProxy,
sessions: Arc<Mutex<HashMap<(Jid, String), Session>>>,
}
impl IbbPlugin {
pub fn new() -> IbbPlugin {
IbbPlugin {
proxy: PluginProxy::new(),
sessions: Arc::new(Mutex::new(HashMap::new())),
}
}
// TODO: make that called automatically after plugins are created.
pub fn init(&self) {
if let Some(disco) = self.proxy.plugin::<DiscoPlugin>() {
disco.add_feature(ns::IBB);
} else {
panic!("Please handle dependencies in the correct order.");
}
}
// TODO: make that called automatically before removal.
pub fn deinit(&self) {
if let Some(disco) = self.proxy.plugin::<DiscoPlugin>() {
disco.remove_feature(ns::IBB);
} else {
panic!("Please handle dependencies in the correct order.");
}
}
fn handle_ibb(&self, from: Jid, ibb: IBB) -> Result<(), StanzaError> {
let mut sessions = self.sessions.lock().unwrap();
match ibb {
IBB::Open { block_size, sid, stanza } => {
match sessions.entry((from.clone(), sid.clone())) {
Entry::Vacant(_) => Ok(()),
Entry::Occupied(_) => Err(generate_error(
ErrorType::Cancel,
DefinedCondition::NotAcceptable,
"This session is already open."
)),
}?;
let session = Session {
stanza,
block_size,
cur_seq: 65535u16,
};
sessions.insert((from, sid), session.clone());
self.proxy.dispatch(IbbOpen {
session: session,
});
},
IBB::Data { seq, sid, data } => {
let entry = match sessions.entry((from, sid)) {
Entry::Occupied(entry) => Ok(entry),
Entry::Vacant(_) => Err(generate_error(
ErrorType::Cancel,
DefinedCondition::ItemNotFound,
"This session doesn’t exist."
)),
}?;
let mut session = entry.into_mut();
if session.stanza != Stanza::Iq {
return Err(generate_error(
ErrorType::Cancel,
DefinedCondition::NotAcceptable,
"Wrong stanza type."
))
}
let cur_seq = session.cur_seq.wrapping_add(1);
if seq != cur_seq {
return Err(generate_error(
ErrorType::Cancel,
DefinedCondition::NotAcceptable,
"Wrong seq number."
))
}
session.cur_seq = cur_seq;
self.proxy.dispatch(IbbData {
session: session.clone(),
data,
});
},
IBB::Close { sid } => {
let entry = match sessions.entry((from, sid)) {
Entry::Occupied(entry) => Ok(entry),
Entry::Vacant(_) => Err(generate_error(
ErrorType::Cancel,
DefinedCondition::ItemNotFound,
"This session doesn’t exist."
)),
}?;
let session = entry.remove();
self.proxy.dispatch(IbbClose {
session,
});
},
}
Ok(())
}
fn handle_iq(&self, iq: &Iq) -> Propagation {
let iq = iq.clone();
if let IqType::Set(payload) = iq.payload {
let from = iq.from.unwrap();
let id = iq.id.unwrap();
// TODO: use an intermediate plugin to parse this payload.
let payload = match IqSetPayload::try_from(payload) {
Ok(IqSetPayload::IBB(ibb)) => {
match self.handle_ibb(from.clone(), ibb) {
Ok(_) => IqType::Result(None),
Err(error) => IqType::Error(error),
}
},
Err(err) => IqType::Error(generate_error(
ErrorType::Cancel,
DefinedCondition::NotAcceptable,
format!("{:?}", err).as_ref()
)),
Ok(_) => return Propagation::Continue,
};
self.proxy.send(Iq {
from: None,
to: Some(from),
id: Some(id),
payload: payload,
}.into());
Propagation::Stop
} else {
Propagation::Continue
}
}
}
impl_plugin!(IbbPlugin, proxy, [
(Iq, Priority::Default) => handle_iq,
]);

132
src/plugins/messaging.rs
View file

@ -1,132 +0,0 @@
use try_from::TryFrom;
use std::collections::BTreeMap;
use plugin::PluginProxy;
use event::{Event, Priority, Propagation};
use error::Error;
use jid::Jid;
use plugins::stanza::Message;
use xmpp_parsers::message::{MessagePayload, MessageType};
use xmpp_parsers::chatstates::ChatState;
use xmpp_parsers::receipts::Receipt;
use xmpp_parsers::stanza_id::StanzaId;
// TODO: use the id (maybe even stanza-id) to identify every message.
#[derive(Debug)]
pub struct MessageEvent {
pub from: Jid,
pub body: String,
pub subject: Option<String>,
pub thread: Option<String>,
}
#[derive(Debug)]
pub struct ChatStateEvent {
pub from: Jid,
pub chat_state: ChatState,
}
#[derive(Debug)]
pub struct ReceiptRequestEvent {
pub from: Jid,
}
#[derive(Debug)]
pub struct ReceiptReceivedEvent {
pub from: Jid,
pub id: String,
}
#[derive(Debug)]
pub struct StanzaIdEvent {
pub from: Jid,
pub stanza_id: StanzaId,
pub message: Message,
}
impl Event for MessageEvent {}
impl Event for ChatStateEvent {}
impl Event for ReceiptRequestEvent {}
impl Event for ReceiptReceivedEvent {}
impl Event for StanzaIdEvent {}
pub struct MessagingPlugin {
proxy: PluginProxy,
}
impl MessagingPlugin {
pub fn new() -> MessagingPlugin {
MessagingPlugin {
proxy: PluginProxy::new(),
}
}
pub fn send_message(&self, to: &Jid, body: &str) -> Result<(), Error> {
let message = Message {
from: None,
to: Some(to.clone()),
type_: MessageType::Chat,
id: Some(self.proxy.gen_id()),
bodies: {
let mut bodies = BTreeMap::new();
bodies.insert(String::new(), String::from(body));
bodies
},
subjects: BTreeMap::new(),
thread: None,
payloads: vec!(),
};
self.proxy.send(message.into());
Ok(())
}
fn handle_message(&self, message: &Message) -> Propagation {
let from = message.from.clone().unwrap_or(self.proxy.get_own_jid());
for payload in message.payloads.clone() {
let payload = match MessagePayload::try_from(payload) {
Ok(payload) => payload,
Err(err) => {
println!("MessagePayload: {:?}", err);
continue;
}
};
match payload {
// XEP-0085
MessagePayload::ChatState(chat_state) => self.proxy.dispatch(ChatStateEvent {
from: from.clone(),
chat_state: chat_state,
}),
// XEP-0184
MessagePayload::Receipt(Receipt::Request) => self.proxy.dispatch(ReceiptRequestEvent {
from: from.clone(),
}),
// XEP-0184
MessagePayload::Receipt(Receipt::Received(id)) => self.proxy.dispatch(ReceiptReceivedEvent {
from: from.clone(),
id: id.unwrap(),
}),
// XEP-0359
MessagePayload::StanzaId(stanza_id) => self.proxy.dispatch(StanzaIdEvent {
from: from.clone(),
stanza_id: stanza_id,
message: message.clone(),
}),
payload => println!("Unhandled payload: {:?}", payload),
}
}
if message.bodies.contains_key("") {
self.proxy.dispatch(MessageEvent {
from: from,
body: message.bodies[""].clone(),
subject: if message.subjects.contains_key("") { Some(message.subjects[""].clone()) } else { None },
thread: message.thread.clone(),
});
}
Propagation::Stop
}
}
impl_plugin!(MessagingPlugin, proxy, [
(Message, Priority::Default) => handle_message,
]);

11
src/plugins/mod.rs
View file

@ -1,11 +0,0 @@
pub mod messaging;
pub mod presence;
pub mod roster;
pub mod disco;
pub mod caps;
pub mod ping;
pub mod ibb;
pub mod stanza;
pub mod stanza_debug;
pub mod unhandled_iq;
pub mod muc;

94
src/plugins/muc.rs
View file

@ -1,94 +0,0 @@
use std::collections::BTreeMap;
use try_from::TryFrom;
use jid::Jid;
use error::Error;
use plugin::PluginProxy;
use event::{Event, Propagation, Priority};
pub use xmpp_parsers::muc::{Muc, MucUser};
pub use xmpp_parsers::muc::user::{Status, Affiliation, Role};
pub use xmpp_parsers::presence::{Presence, Type, Show};
#[derive(Debug)]
pub struct MucPresence {
pub room: Jid,
pub nick: Option<String>,
pub to: Jid,
pub type_: Type,
pub x: MucUser,
}
impl Event for MucPresence {}
pub struct MucPlugin {
proxy: PluginProxy,
}
impl MucPlugin {
pub fn new() -> MucPlugin {
MucPlugin {
proxy: PluginProxy::new(),
}
}
pub fn join_room(&self, room: Jid) -> Result<(), Error> {
let x = Muc { password: None };
let presence = Presence {
from: None,
to: Some(room),
id: None,
type_: Type::None,
show: Show::None,
priority: 0i8,
statuses: BTreeMap::new(),
payloads: vec![x.into()],
};
self.proxy.send(presence.into());
Ok(())
}
pub fn leave_room(&self, room: Jid) -> Result<(), Error> {
let x = Muc { password: None };
let presence = Presence {
from: None,
to: Some(room),
id: None,
type_: Type::None,
show: Show::None,
priority: 0i8,
statuses: BTreeMap::new(),
payloads: vec![x.into()],
};
self.proxy.send(presence.into());
Ok(())
}
fn handle_presence(&self, presence: &Presence) -> Propagation {
let from = presence.from.clone().unwrap();
let room = from.clone().into_bare_jid();
let nick = from.resource;
let to = presence.to.clone().unwrap();
let type_ = presence.type_.clone();
for payload in presence.clone().payloads {
if let Ok(x) = MucUser::try_from(payload) {
self.proxy.dispatch(MucPresence {
room: room.clone(),
nick: nick.clone(),
to: to.clone(),
type_: type_.clone(),
x
});
}
}
Propagation::Stop
}
}
impl_plugin!(MucPlugin, proxy, [
(Presence, Priority::Default) => handle_presence,
]);

91
src/plugins/ping.rs
View file

@ -1,91 +0,0 @@
use try_from::TryFrom;
use plugin::PluginProxy;
use event::{Event, Priority, Propagation};
use error::Error;
use jid::Jid;
use plugins::stanza::Iq;
use plugins::disco::DiscoPlugin;
use xmpp_parsers::iq::{IqType, IqGetPayload};
use xmpp_parsers::ping::Ping;
use xmpp_parsers::ns;
#[derive(Debug)]
pub struct PingEvent {
pub from: Jid,
pub id: String,
}
impl Event for PingEvent {}
pub struct PingPlugin {
proxy: PluginProxy,
}
impl PingPlugin {
pub fn new() -> PingPlugin {
PingPlugin {
proxy: PluginProxy::new(),
}
}
// TODO: make that called automatically after plugins are created.
pub fn init(&self) {
if let Some(disco) = self.proxy.plugin::<DiscoPlugin>() {
disco.add_feature(ns::PING);
} else {
panic!("Please handle dependencies in the correct order.");
}
}
// TODO: make that called automatically before removal.
pub fn deinit(&self) {
if let Some(disco) = self.proxy.plugin::<DiscoPlugin>() {
disco.remove_feature(ns::PING);
} else {
panic!("Please handle dependencies in the correct order.");
}
}
pub fn send_ping(&self, to: &Jid) -> Result<(), Error> {
let to = to.clone();
self.proxy.send(Iq {
from: None,
to: Some(to),
id: Some(self.proxy.gen_id()),
payload: IqType::Get(IqGetPayload::Ping(Ping).into()),
}.into());
Ok(())
}
fn handle_iq(&self, iq: &Iq) -> Propagation {
let iq = iq.clone();
if let IqType::Get(payload) = iq.payload {
// TODO: use an intermediate plugin to parse this payload.
if let Ok(IqGetPayload::Ping(_)) = IqGetPayload::try_from(payload) {
self.proxy.dispatch(PingEvent { // TODO: safety!!!
from: iq.from.unwrap(),
id: iq.id.unwrap(),
});
return Propagation::Stop;
}
}
Propagation::Continue
}
fn reply_ping(&self, ping: &PingEvent) -> Propagation {
self.proxy.send(Iq {
from: None,
to: Some(ping.from.to_owned()),
id: Some(ping.id.to_owned()),
payload: IqType::Result(None),
}.into());
Propagation::Continue
}
}
impl_plugin!(PingPlugin, proxy, [
(Iq, Priority::Default) => handle_iq,
(PingEvent, Priority::Default) => reply_ping,
]);

41
src/plugins/presence.rs
View file

@ -1,41 +0,0 @@
use std::collections::BTreeMap;
use error::Error;
use plugin::PluginProxy;
pub use xmpp_parsers::presence::{Presence, Type, Show};
pub struct PresencePlugin {
proxy: PluginProxy,
}
impl PresencePlugin {
pub fn new() -> PresencePlugin {
PresencePlugin {
proxy: PluginProxy::new(),
}
}
pub fn set_presence(&self, type_: Type, show: Show, status: Option<String>) -> Result<(), Error> {
let presence = Presence {
from: None,
to: None,
id: Some(self.proxy.gen_id()),
type_: type_,
show: show,
priority: 0i8,
statuses: {
let mut statuses = BTreeMap::new();
if let Some(status) = status {
statuses.insert(String::new(), status);
}
statuses
},
payloads: vec!(),
};
self.proxy.send(presence.into());
Ok(())
}
}
impl_plugin!(PresencePlugin, proxy, []);

186
src/plugins/roster.rs
View file

@ -1,186 +0,0 @@
use std::collections::HashMap;
use try_from::TryFrom;
use std::sync::Mutex;
use plugin::PluginProxy;
use event::{Event, Priority, Propagation};
use jid::Jid;
use plugins::stanza::Iq;
use plugins::disco::DiscoPlugin;
use xmpp_parsers::iq::{IqType, IqSetPayload, IqResultPayload};
use xmpp_parsers::roster::{Roster, Item, Subscription};
use xmpp_parsers::ns;
#[derive(Debug)]
pub struct RosterReceived {
pub ver: Option<String>,
pub jids: HashMap<Jid, Item>,
}
#[derive(Debug)]
pub enum RosterPush {
Added(Item),
Modified(Item),
Removed(Item),
}
impl Event for RosterReceived {}
impl Event for RosterPush {}
pub struct RosterPlugin {
proxy: PluginProxy,
current_version: Mutex<Option<String>>,
// TODO: allow for a different backing store.
jids: Mutex<HashMap<Jid, Item>>,
}
impl RosterPlugin {
pub fn new(ver: Option<String>) -> RosterPlugin {
RosterPlugin {
proxy: PluginProxy::new(),
current_version: Mutex::new(ver),
jids: Mutex::new(HashMap::new()),
}
}
// TODO: make that called automatically after plugins are created.
pub fn init(&self) {
if let Some(disco) = self.proxy.plugin::<DiscoPlugin>() {
disco.add_feature(ns::IBB);
} else {
panic!("Please handle dependencies in the correct order.");
}
}
// TODO: make that called automatically before removal.
pub fn deinit(&self) {
if let Some(disco) = self.proxy.plugin::<DiscoPlugin>() {
disco.remove_feature(ns::IBB);
} else {
panic!("Please handle dependencies in the correct order.");
}
}
pub fn send_roster_get(&self, ver: Option<String>) {
let iq = Iq {
from: None,
to: None,
id: Some(self.proxy.gen_id()),
payload: IqType::Get(Roster {
ver,
items: vec!(),
}.into()),
};
self.proxy.send(iq.into());
}
// TODO: use a better error type.
pub fn send_roster_set(&self, to: Option<Jid>, item: Item) -> Result<(), String> {
if item.subscription.is_some() && item.subscription != Some(Subscription::Remove) {
return Err(String::from("Subscription must be either nothing or Remove."));
}
let iq = Iq {
from: None,
to,
id: Some(self.proxy.gen_id()),
payload: IqType::Set(Roster {
ver: None,
items: vec!(item),
}.into()),
};
self.proxy.send(iq.into());
Ok(())
}
fn handle_roster_reply(&self, roster: Roster) {
// TODO: handle the same-ver case!
let mut current_version = self.current_version.lock().unwrap();
*current_version = roster.ver;
let mut jids = self.jids.lock().unwrap();
jids.clear();
for item in roster.items {
jids.insert(item.jid.clone(), item);
}
self.proxy.dispatch(RosterReceived {
ver: current_version.clone(),
jids: jids.clone(),
});
}
fn handle_roster_push(&self, roster: Roster) -> Result<(), String> {
let item = roster.items.get(0);
if item.is_none() || roster.items.len() != 1 {
return Err(String::from("Server sent an invalid roster push!"));
}
let item = item.unwrap().clone();
let mut jids = self.jids.lock().unwrap();
let previous = jids.insert(item.jid.clone(), item.clone());
if previous.is_none() {
assert!(item.subscription != Some(Subscription::Remove));
self.proxy.dispatch(RosterPush::Added(item));
} else {
if item.subscription == Some(Subscription::Remove) {
self.proxy.dispatch(RosterPush::Removed(item));
} else {
self.proxy.dispatch(RosterPush::Modified(item));
}
}
Ok(())
}
fn handle_iq(&self, iq: &Iq) -> Propagation {
let jid = self.proxy.get_own_jid();
let jid = Jid::bare(jid.node.unwrap(), jid.domain);
if iq.from.is_some() && iq.from != Some(jid) {
// Not from our roster.
return Propagation::Continue;
}
let iq = iq.clone();
let id = iq.id.unwrap();
match iq.payload {
IqType::Result(Some(payload)) => {
match IqResultPayload::try_from(payload) {
Ok(IqResultPayload::Roster(roster)) => {
self.handle_roster_reply(roster);
Propagation::Stop
},
Ok(_)
| Err(_) => Propagation::Continue,
}
},
IqType::Set(payload) => {
match IqSetPayload::try_from(payload) {
Ok(IqSetPayload::Roster(roster)) => {
let payload = match self.handle_roster_push(roster) {
Ok(_) => IqType::Result(None),
Err(string) => {
// The specification says that the server should ignore an error.
println!("{}", string);
IqType::Result(None)
},
};
self.proxy.send(Iq {
from: None,
to: None,
id: Some(id),
payload: payload,
}.into());
Propagation::Stop
},
Ok(_)
| Err(_) => return Propagation::Continue,
}
},
IqType::Result(None)
| IqType::Get(_)
| IqType::Error(_) => {
Propagation::Continue
},
}
}
}
impl_plugin!(RosterPlugin, proxy, [
(Iq, Priority::Default) => handle_iq,
]);

52
src/plugins/stanza.rs
View file

@ -1,52 +0,0 @@
use try_from::TryFrom;
use plugin::PluginProxy;
use event::{Event, ReceiveElement, Propagation, Priority};
use ns;
pub use xmpp_parsers::message::Message;
pub use xmpp_parsers::presence::Presence;
pub use xmpp_parsers::iq::Iq;
impl Event for Message {}
impl Event for Presence {}
impl Event for Iq {}
pub struct StanzaPlugin {
proxy: PluginProxy,
}
impl StanzaPlugin {
pub fn new() -> StanzaPlugin {
StanzaPlugin {
proxy: PluginProxy::new(),
}
}
fn handle_receive_element(&self, evt: &ReceiveElement) -> Propagation {
let elem = &evt.0;
// TODO: make the handle take an Element instead of a reference.
let elem = elem.clone();
if elem.is("message", ns::CLIENT) {
let message = Message::try_from(elem).unwrap();
self.proxy.dispatch(message);
} else if elem.is("presence", ns::CLIENT) {
let presence = Presence::try_from(elem).unwrap();
self.proxy.dispatch(presence);
} else if elem.is("iq", ns::CLIENT) {
let iq = Iq::try_from(elem).unwrap();
self.proxy.dispatch(iq);
} else {
// TODO: handle nonzas too.
return Propagation::Continue;
}
Propagation::Stop
}
}
impl_plugin!(StanzaPlugin, proxy, [
(ReceiveElement, Priority::Default) => handle_receive_element,
]);

30
src/plugins/stanza_debug.rs
View file

@ -1,30 +0,0 @@
use plugin::PluginProxy;
use event::{SendElement, ReceiveElement, Propagation, Priority};
use chrono::Local;
pub struct StanzaDebugPlugin {
proxy: PluginProxy,
}
impl StanzaDebugPlugin {
pub fn new() -> StanzaDebugPlugin {
StanzaDebugPlugin {
proxy: PluginProxy::new(),
}
}
fn handle_send_element(&self, evt: &SendElement) -> Propagation {
println!("{} SEND: {:?}", Local::now(), evt.0);
Propagation::Continue
}
fn handle_receive_element(&self, evt: &ReceiveElement) -> Propagation {
println!("{} RECV: {:?}", Local::now(), evt.0);
Propagation::Continue
}
}
impl_plugin!(StanzaDebugPlugin, proxy, [
(SendElement, Priority::Min) => handle_send_element,
(ReceiveElement, Priority::Max) => handle_receive_element,
]);

48
src/plugins/unhandled_iq.rs
View file

@ -1,48 +0,0 @@
use std::collections::BTreeMap;
use plugin::PluginProxy;
use event::{Priority, Propagation};
use plugins::stanza::Iq;
use xmpp_parsers::iq::IqType;
use xmpp_parsers::stanza_error::{StanzaError, ErrorType, DefinedCondition};
pub struct UnhandledIqPlugin {
proxy: PluginProxy,
}
impl UnhandledIqPlugin {
pub fn new() -> UnhandledIqPlugin {
UnhandledIqPlugin {
proxy: PluginProxy::new(),
}
}
fn reply_unhandled_iq(&self, iq: &Iq) -> Propagation {
let iq = iq.clone();
match iq.payload {
IqType::Get(_)
| IqType::Set(_) => {
self.proxy.send(Iq {
from: None,
to: Some(iq.from.unwrap()),
id: Some(iq.id.unwrap()),
payload: IqType::Error(StanzaError {
type_: ErrorType::Cancel,
defined_condition: DefinedCondition::ServiceUnavailable,
texts: BTreeMap::new(),
by: None,
other: None,
}),
}.into());
Propagation::Stop
},
IqType::Result(_)
| IqType::Error(_) => Propagation::Continue
}
}
}
impl_plugin!(UnhandledIqPlugin, proxy, [
(Iq, Priority::Min) => reply_unhandled_iq,
]);

227
src/transport.rs
View file

@ -1,227 +0,0 @@
//! Provides transports for the xml streams.
use std::io::prelude::*;
use std::net::{TcpStream, Shutdown};
use xml::reader::{EventReader, XmlEvent as XmlReaderEvent};
use xml::writer::{EventWriter, XmlEvent as XmlWriterEvent, EmitterConfig};
use std::sync::{Arc, Mutex};
use ns;
use minidom;
use locked_io::LockedIO;
use error::Error;
#[allow(unused_imports)]
use openssl::ssl::{SslMethod, Ssl, SslContextBuilder, SslStream, SSL_VERIFY_NONE, SslConnectorBuilder};
use sasl::common::ChannelBinding;
/// A trait which transports are required to implement.
pub trait Transport {
/// Writes an `xml::writer::XmlEvent` to the stream.
fn write_event<'a, E: Into<XmlWriterEvent<'a>>>(&mut self, event: E) -> Result<(), Error>;
/// Reads an `xml::reader::XmlEvent` from the stream.
fn read_event(&mut self) -> Result<XmlReaderEvent, Error>;
/// Writes a `minidom::Element` to the stream.
fn write_element(&mut self, element: &minidom::Element) -> Result<(), Error>;
/// Reads a `minidom::Element` from the stream.
fn read_element(&mut self) -> Result<minidom::Element, Error>;
/// Resets the stream.
fn reset_stream(&mut self);
/// Gets channel binding data.
fn channel_bind(&self) -> ChannelBinding {
ChannelBinding::None
}
}
/// A plain text transport, completely unencrypted.
pub struct PlainTransport {
inner: Arc<Mutex<TcpStream>>, // TODO: this feels rather ugly
reader: EventReader<LockedIO<TcpStream>>, // TODO: especially feels ugly because
// this read would keep the lock
// held very long (potentially)
writer: EventWriter<LockedIO<TcpStream>>,
}
impl Transport for PlainTransport {
fn write_event<'a, E: Into<XmlWriterEvent<'a>>>(&mut self, event: E) -> Result<(), Error> {
Ok(self.writer.write(event)?)
}
fn read_event(&mut self) -> Result<XmlReaderEvent, Error> {
Ok(self.reader.next()?)
}
fn write_element(&mut self, element: &minidom::Element) -> Result<(), Error> {
Ok(element.write_to(&mut self.writer)?)
}
fn read_element(&mut self) -> Result<minidom::Element, Error> {
let element = minidom::Element::from_reader(&mut self.reader)?;
Ok(element)
}
fn reset_stream(&mut self) {
let locked_io = LockedIO::from(self.inner.clone());
self.reader = EventReader::new(locked_io.clone());
self.writer = EventWriter::new_with_config(locked_io, EmitterConfig {
line_separator: "".into(),
perform_indent: false,
normalize_empty_elements: false,
.. Default::default()
});
}
fn channel_bind(&self) -> ChannelBinding {
// TODO: channel binding
ChannelBinding::None
}
}
impl PlainTransport {
/// Connects to a server without any encryption.
pub fn connect(host: &str, port: u16) -> Result<PlainTransport, Error> {
let tcp_stream = TcpStream::connect((host, port))?;
let parser = EventReader::new(tcp_stream);
let parser_stream = parser.into_inner();
let stream = Arc::new(Mutex::new(parser_stream));
let locked_io = LockedIO::from(stream.clone());
let reader = EventReader::new(locked_io.clone());
let writer = EventWriter::new_with_config(locked_io, EmitterConfig {
line_separator: "".into(),
perform_indent: false,
normalize_empty_elements: false,
.. Default::default()
});
Ok(PlainTransport {
inner: stream,
reader: reader,
writer: writer,
})
}
/// Closes the stream.
pub fn close(&mut self) {
self.inner.lock()
.unwrap()
.shutdown(Shutdown::Both)
.unwrap(); // TODO: safety, return value and such
}
}
/// A transport which uses STARTTLS.
pub struct SslTransport {
inner: Arc<Mutex<SslStream<TcpStream>>>, // TODO: this feels rather ugly
reader: EventReader<LockedIO<SslStream<TcpStream>>>, // TODO: especially feels ugly because
// this read would keep the lock
// held very long (potentially)
writer: EventWriter<LockedIO<SslStream<TcpStream>>>,
}
impl Transport for SslTransport {
fn write_event<'a, E: Into<XmlWriterEvent<'a>>>(&mut self, event: E) -> Result<(), Error> {
Ok(self.writer.write(event)?)
}
fn read_event(&mut self) -> Result<XmlReaderEvent, Error> {
Ok(self.reader.next()?)
}
fn write_element(&mut self, element: &minidom::Element) -> Result<(), Error> {
Ok(element.write_to(&mut self.writer)?)
}
fn read_element(&mut self) -> Result<minidom::Element, Error> {
Ok(minidom::Element::from_reader(&mut self.reader)?)
}
fn reset_stream(&mut self) {
let locked_io = LockedIO::from(self.inner.clone());
self.reader = EventReader::new(locked_io.clone());
self.writer = EventWriter::new_with_config(locked_io, EmitterConfig {
line_separator: "".into(),
perform_indent: false,
normalize_empty_elements: false,
.. Default::default()
});
}
fn channel_bind(&self) -> ChannelBinding {
// TODO: channel binding
ChannelBinding::None
}
}
impl SslTransport {
/// Connects to a server using STARTTLS.
pub fn connect(host: &str, port: u16) -> Result<SslTransport, Error> {
// TODO: very quick and dirty, blame starttls
let mut stream = TcpStream::connect((host, port))?;
write!(stream, "<stream:stream xmlns='{}' xmlns:stream='{}' to='{}' version='1.0'>"
, ns::CLIENT, ns::STREAM, host)?;
write!(stream, "<starttls xmlns='{}'/>"
, ns::TLS)?;
let mut parser = EventReader::new(stream);
loop { // TODO: possibly a timeout?
match parser.next()? {
XmlReaderEvent::StartElement { name, .. } => {
if let Some(ns) = name.namespace {
if ns == ns::TLS && name.local_name == "proceed" {
break;
}
else if ns == ns::STREAM && name.local_name == "error" {
return Err(Error::StreamError);
}
}
},
_ => {},
}
}
let stream = parser.into_inner();
#[cfg(feature = "insecure")]
let ssl_stream = {
let mut ctx = SslContextBuilder::new(SslMethod::tls())?;
ctx.set_verify(SSL_VERIFY_NONE);
let ssl = Ssl::new(&ctx.build())?;
ssl.connect(stream)?
};
#[cfg(not(feature = "insecure"))]
let ssl_stream = {
let ssl_connector = SslConnectorBuilder::new(SslMethod::tls())?.build();
ssl_connector.connect(host, stream)?
};
let ssl_stream = Arc::new(Mutex::new(ssl_stream));
let locked_io = LockedIO::from(ssl_stream.clone());
let reader = EventReader::new(locked_io.clone());
let writer = EventWriter::new_with_config(locked_io, EmitterConfig {
line_separator: "".into(),
perform_indent: false,
normalize_empty_elements: false,
.. Default::default()
});
Ok(SslTransport {
inner: ssl_stream,
reader: reader,
writer: writer,
})
}
/// Closes the stream.
pub fn close(&mut self) {
self.inner.lock()
.unwrap()
.shutdown()
.unwrap(); // TODO: safety, return value and such
}
}

19
src/util.rs
View file

@ -1,19 +0,0 @@
use minidom::Element;
pub trait FromElement where Self: Sized {
type Err;
fn from_element(elem: &Element) -> Result<Self, Self::Err>;
}
pub trait FromParentElement where Self: Sized {
type Err;
fn from_parent_element(elem: &Element) -> Result<Self, Self::Err>;
}
pub trait ToElement where Self: Sized {
type Err;
fn to_element(&self) -> Result<Element, Self::Err>;
}