1// Copyright (c) 2019 Emmanuel Gil Peyrot <linkmauve@linkmauve.fr>
2//
3// This Source Code Form is subject to the terms of the Mozilla Public
4// License, v. 2.0. If a copy of the MPL was not distributed with this
5// file, You can obtain one at http://mozilla.org/MPL/2.0/.
6
7#![deny(bare_trait_objects)]
8
9use std::str::FromStr;
10use std::rc::Rc;
11use std::cell::RefCell;
12use std::convert::TryFrom;
13use futures::{Future,Stream, Sink, sync::mpsc};
14use tokio_xmpp::{
15 Client as TokioXmppClient,
16 Event as TokioXmppEvent,
17 Packet,
18};
19use xmpp_parsers::{
20 bookmarks::{
21 Autojoin,
22 Conference as ConferenceBookmark,
23 Storage as Bookmarks,
24 },
25 caps::{compute_disco, hash_caps, Caps},
26 disco::{DiscoInfoQuery, DiscoInfoResult, Feature, Identity},
27 hashes::Algo,
28 iq::{Iq, IqType},
29 message::{Message, MessageType, Body},
30 muc::{
31 Muc,
32 user::{MucUser, Status},
33 },
34 ns,
35 presence::{Presence, Type as PresenceType},
36 pubsub::{
37 event::PubSubEvent,
38 pubsub::PubSub,
39 },
40 roster::{Roster, Item as RosterItem},
41 stanza_error::{StanzaError, ErrorType, DefinedCondition},
42 Jid, BareJid, FullJid, JidParseError,
43};
44
45mod avatar;
46
47pub type Error = tokio_xmpp::Error;
48
49#[derive(Debug)]
50pub enum ClientType {
51 Bot,
52 Pc,
53}
54
55impl Default for ClientType {
56 fn default() -> Self {
57 ClientType::Bot
58 }
59}
60
61impl ToString for ClientType {
62 fn to_string(&self) -> String {
63 String::from(
64 match self {
65 ClientType::Bot => "bot",
66 ClientType::Pc => "pc",
67 }
68 )
69 }
70}
71
72#[derive(PartialEq)]
73pub enum ClientFeature {
74 Avatars,
75 ContactList,
76 JoinRooms,
77}
78
79#[derive(Debug)]
80pub enum Event {
81 Online,
82 Disconnected,
83 ContactAdded(RosterItem),
84 ContactRemoved(RosterItem),
85 ContactChanged(RosterItem),
86 AvatarRetrieved(Jid, String),
87 OpenRoomBookmark(ConferenceBookmark),
88 RoomJoined(BareJid),
89 RoomLeft(BareJid),
90}
91
92#[derive(Default)]
93pub struct ClientBuilder<'a> {
94 jid: &'a str,
95 password: &'a str,
96 website: String,
97 default_nick: String,
98 disco: (ClientType, String),
99 features: Vec<ClientFeature>,
100}
101
102impl ClientBuilder<'_> {
103 pub fn new<'a>(jid: &'a str, password: &'a str) -> ClientBuilder<'a> {
104 ClientBuilder {
105 jid,
106 password,
107 website: String::from("https://gitlab.com/xmpp-rs/tokio-xmpp"),
108 default_nick: String::from("xmpp-rs"),
109 disco: (ClientType::default(), String::from("tokio-xmpp")),
110 features: vec![],
111 }
112 }
113
114 pub fn set_client(mut self, type_: ClientType, name: &str) -> Self {
115 self.disco = (type_, String::from(name));
116 self
117 }
118
119 pub fn set_website(mut self, url: &str) -> Self {
120 self.website = String::from(url);
121 self
122 }
123
124 pub fn set_default_nick(mut self, nick: &str) -> Self {
125 self.default_nick = String::from(nick);
126 self
127 }
128
129 pub fn enable_feature(mut self, feature: ClientFeature) -> Self {
130 self.features.push(feature);
131 self
132 }
133
134 fn make_disco(&self) -> DiscoInfoResult {
135 let identities = vec![Identity::new("client", self.disco.0.to_string(),
136 "en", self.disco.1.to_string())];
137 let mut features = vec![
138 Feature::new(ns::DISCO_INFO),
139 ];
140 if self.features.contains(&ClientFeature::Avatars) {
141 features.push(Feature::new(format!("{}+notify", ns::AVATAR_METADATA)));
142 }
143 if self.features.contains(&ClientFeature::JoinRooms) {
144 features.push(Feature::new(format!("{}+notify", ns::BOOKMARKS)));
145 }
146 DiscoInfoResult {
147 node: None,
148 identities,
149 features,
150 extensions: vec![],
151 }
152 }
153
154 fn make_initial_presence(disco: &DiscoInfoResult, node: &str) -> Presence {
155 let caps_data = compute_disco(disco);
156 let hash = hash_caps(&caps_data, Algo::Sha_1).unwrap();
157 let caps = Caps::new(node, hash);
158
159 let mut presence = Presence::new(PresenceType::None);
160 presence.add_payload(caps);
161 presence
162 }
163
164 pub fn build(
165 self,
166 ) -> Result<(Agent, impl Stream<Item = Event, Error = tokio_xmpp::Error>), JidParseError> {
167 let client = TokioXmppClient::new(self.jid, self.password)?;
168 Ok(self.build_impl(client))
169 }
170
171 // This function is meant to be used for testing build
172 pub(crate) fn build_impl<S>(
173 self,
174 stream: S,
175 ) -> (Agent, impl Stream<Item = Event, Error = tokio_xmpp::Error>)
176 where
177 S: Stream<Item = tokio_xmpp::Event, Error = tokio_xmpp::Error>
178 + Sink<SinkItem = tokio_xmpp::Packet, SinkError = tokio_xmpp::Error>,
179 {
180 let disco = self.make_disco();
181 let node = self.website;
182 let (sender_tx, sender_rx) = mpsc::unbounded();
183
184 let client = stream;
185 let (sink, stream) = client.split();
186
187 let reader = {
188 let mut sender_tx = sender_tx.clone();
189 let jid = self.jid.to_owned();
190 stream.map(move |event| {
191 // Helper function to send an iq error.
192 let mut events = Vec::new();
193 let send_error = |to, id, type_, condition, text: &str| {
194 let error = StanzaError::new(type_, condition, "en", text);
195 let iq = Iq::from_error(id, error)
196 .with_to(to)
197 .into();
198 sender_tx.unbounded_send(Packet::Stanza(iq)).unwrap();
199 };
200
201 match event {
202 TokioXmppEvent::Online => {
203 let presence = ClientBuilder::make_initial_presence(&disco, &node).into();
204 let packet = Packet::Stanza(presence);
205 sender_tx.unbounded_send(packet)
206 .unwrap();
207 events.push(Event::Online);
208 // TODO: only send this when the ContactList feature is enabled.
209 let iq = Iq::from_get("roster", Roster { ver: None, items: vec![] })
210 .into();
211 sender_tx.unbounded_send(Packet::Stanza(iq)).unwrap();
212 }
213 TokioXmppEvent::Disconnected => {
214 events.push(Event::Disconnected);
215 }
216 TokioXmppEvent::Stanza(stanza) => {
217 if stanza.is("iq", "jabber:client") {
218 let iq = Iq::try_from(stanza).unwrap();
219 if let IqType::Get(payload) = iq.payload {
220 if payload.is("query", ns::DISCO_INFO) {
221 let query = DiscoInfoQuery::try_from(payload);
222 match query {
223 Ok(query) => {
224 let mut disco_info = disco.clone();
225 disco_info.node = query.node;
226 let iq = Iq::from_result(iq.id, Some(disco_info))
227 .with_to(iq.from.unwrap())
228 .into();
229 sender_tx.unbounded_send(Packet::Stanza(iq)).unwrap();
230 },
231 Err(err) => {
232 send_error(iq.from.unwrap(), iq.id, ErrorType::Modify, DefinedCondition::BadRequest, &format!("{}", err));
233 },
234 }
235 } else {
236 // We MUST answer unhandled get iqs with a service-unavailable error.
237 send_error(iq.from.unwrap(), iq.id, ErrorType::Cancel, DefinedCondition::ServiceUnavailable, "No handler defined for this kind of iq.");
238 }
239 } else if let IqType::Result(Some(payload)) = iq.payload {
240 // TODO: move private iqs like this one somewhere else, for
241 // security reasons.
242 if payload.is("query", ns::ROSTER) && iq.from.is_none() {
243 let roster = Roster::try_from(payload).unwrap();
244 for item in roster.items.into_iter() {
245 events.push(Event::ContactAdded(item));
246 }
247 } else if payload.is("pubsub", ns::PUBSUB) {
248 let pubsub = PubSub::try_from(payload).unwrap();
249 let from =
250 iq.from.clone().unwrap_or_else(|| Jid::from_str(&jid).unwrap());
251 if let PubSub::Items(items) = pubsub {
252 if items.node.0 == ns::AVATAR_DATA {
253 let new_events = avatar::handle_data_pubsub_iq(&from, &items);
254 events.extend(new_events);
255 }
256 }
257 }
258 } else if let IqType::Set(_) = iq.payload {
259 // We MUST answer unhandled set iqs with a service-unavailable error.
260 send_error(iq.from.unwrap(), iq.id, ErrorType::Cancel, DefinedCondition::ServiceUnavailable, "No handler defined for this kind of iq.");
261 }
262 } else if stanza.is("message", "jabber:client") {
263 let message = Message::try_from(stanza).unwrap();
264 let from = message.from.clone().unwrap();
265 for child in message.payloads {
266 if child.is("event", ns::PUBSUB_EVENT) {
267 let event = PubSubEvent::try_from(child).unwrap();
268 if let PubSubEvent::PublishedItems { node, items } = event {
269 if node.0 == ns::AVATAR_METADATA {
270 let new_events = avatar::handle_metadata_pubsub_event(&from, &mut sender_tx, items);
271 events.extend(new_events);
272 } else if node.0 == ns::BOOKMARKS {
273 // TODO: Check that our bare JID is the sender.
274 assert_eq!(items.len(), 1);
275 let item = items.clone().pop().unwrap();
276 let payload = item.payload.clone().unwrap();
277 let bookmarks = match Bookmarks::try_from(payload) {
278 Ok(bookmarks) => bookmarks,
279 // XXX: Don’t panic…
280 Err(err) => panic!("… {}", err),
281 };
282 for bookmark in bookmarks.conferences {
283 if bookmark.autojoin == Autojoin::True {
284 events.push(Event::OpenRoomBookmark(bookmark));
285 }
286 }
287 }
288 }
289 }
290 }
291 } else if stanza.is("presence", "jabber:client") {
292 let presence = Presence::try_from(stanza).unwrap();
293 let from: BareJid = match presence.from.clone().unwrap() {
294 Jid::Full(FullJid { node, domain, .. }) => BareJid { node, domain },
295 Jid::Bare(bare) => bare,
296 };
297 for payload in presence.payloads.into_iter() {
298 let muc_user = match MucUser::try_from(payload) {
299 Ok(muc_user) => muc_user,
300 _ => continue
301 };
302 for status in muc_user.status.into_iter() {
303 if status == Status::SelfPresence {
304 events.push(Event::RoomJoined(from.clone()));
305 break;
306 }
307 }
308 }
309 } else if stanza.is("error", "http://etherx.jabber.org/streams") {
310 println!("Received a fatal stream error: {}", String::from(&stanza));
311 } else {
312 panic!("Unknown stanza: {}", String::from(&stanza));
313 }
314 }
315 }
316
317 futures::stream::iter_ok(events)
318 })
319 .flatten()
320 };
321
322 let sender = sender_rx
323 .map_err(|e| panic!("Sink error: {:?}", e))
324 .forward(sink)
325 .map(|(rx, mut sink)| {
326 drop(rx);
327 let _ = sink.close();
328 None
329 });
330
331 // TODO is this correct?
332 // Some(Error) means a real error
333 // None means the end of the sender stream and can be ignored
334 let future = reader
335 .map(Some)
336 .select(sender.into_stream())
337 .filter_map(|x| x);
338
339 let agent = Agent {
340 sender_tx,
341 default_nick: Rc::new(RefCell::new(self.default_nick)),
342 };
343
344 (agent, future)
345 }
346}
347
348#[derive(Clone)]
349pub struct Agent {
350 sender_tx: mpsc::UnboundedSender<Packet>,
351 default_nick: Rc<RefCell<String>>,
352}
353
354impl Agent {
355 pub fn join_room(&mut self, room: BareJid, nick: Option<String>, password: Option<String>,
356 lang: &str, status: &str) {
357 let mut muc = Muc::new();
358 if let Some(password) = password {
359 muc = muc.with_password(password);
360 }
361
362 let nick = nick.unwrap_or_else(|| self.default_nick.borrow().clone());
363 let room_jid = room.with_resource(nick);
364 let mut presence = Presence::new(PresenceType::None)
365 .with_to(Some(Jid::Full(room_jid)));
366 presence.add_payload(muc);
367 presence.set_status(String::from(lang), String::from(status));
368 let presence = presence.into();
369 self.sender_tx.unbounded_send(Packet::Stanza(presence))
370 .unwrap();
371 }
372
373 pub fn send_message(&mut self, recipient: Jid, type_: MessageType, lang: &str, text: &str) {
374 let mut message = Message::new(Some(recipient));
375 message.type_ = type_;
376 message.bodies.insert(String::from(lang), Body(String::from(text)));
377 let message = message.into();
378 self.sender_tx.unbounded_send(Packet::Stanza(message))
379 .unwrap();
380 }
381}