1#!/usr/bin/env ruby
2#
3# Copyright (C) 2017 Denver Gingerich <denver@ossguy.com>
4# Copyright (C) 2017 Stephen Paul Weber <singpolyma@singpolyma.net>
5#
6# This file is part of sgx-catapult.
7#
8# sgx-catapult is free software: you can redistribute it and/or modify it under
9# the terms of the GNU Affero General Public License as published by the Free
10# Software Foundation, either version 3 of the License, or (at your option) any
11# later version.
12#
13# sgx-catapult is distributed in the hope that it will be useful, but WITHOUT
14# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
15# FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more
16# details.
17#
18# You should have received a copy of the GNU Affero General Public License along
19# with sgx-catapult. If not, see <http://www.gnu.org/licenses/>.
20
21require 'blather/client/dsl'
22require 'em-hiredis'
23require 'em-http-request'
24require 'json'
25require 'securerandom'
26require 'time'
27require 'uri'
28require 'webrick'
29
30require 'goliath/api'
31require 'goliath/server'
32require 'log4r'
33
34require_relative 'em_promise'
35
36def panic(e)
37 puts "Shutting down gateway due to exception: #{e.message}"
38 puts e.backtrace
39 SGXcatapult.shutdown
40 puts 'Gateway has terminated.'
41 EM.stop
42end
43
44def extract_shortcode(dest)
45 num, context = dest.split(';', 2)
46 num if context && context == 'phone-context=ca-us.phone-context.soprani.ca'
47end
48
49class SGXClient < Blather::Client
50 def register_handler(type, *guards, &block)
51 super(type, *guards) do |stanza|
52 begin
53 v = block.call(stanza)
54 v.catch(&method(:panic)) if v.is_a?(Promise)
55 true # Do not run other handlers unless throw :pass
56 rescue Exception => e
57 panic(e)
58 end
59 end
60 end
61end
62
63module SGXcatapult
64 extend Blather::DSL
65
66 @jingle_sids = {}
67 @jingle_fnames = {}
68 @partial_data = {}
69 @client = SGXClient.new
70
71 def self.run
72 client.run
73 end
74
75 # so classes outside this module can write messages, too
76 def self.write(stanza)
77 client.write(stanza)
78 end
79
80 def self.error_msg(orig, query_node, type, name, text=nil)
81 if not query_node.nil?
82 orig.add_child(query_node)
83 orig.type = :error
84 end
85
86 error = Nokogiri::XML::Node.new 'error', orig.document
87 error['type'] = type
88 orig.add_child(error)
89
90 suberr = Nokogiri::XML::Node.new name, orig.document
91 suberr['xmlns'] = 'urn:ietf:params:xml:ns:xmpp-stanzas'
92 error.add_child(suberr)
93
94 # TODO: add some explanatory xml:lang='en' text (see text param)
95 puts "RESPONSE3: #{orig.inspect}"
96 return orig
97 end
98
99 # workqueue_count MUST be 0 or else Blather uses threads!
100 setup ARGV[0], ARGV[1], ARGV[2], ARGV[3], nil, nil, workqueue_count: 0
101
102 def self.pass_on_message(m, users_num, jid)
103 # setup delivery receipt; similar to a reply
104 rcpt = ReceiptMessage.new(m.from.stripped)
105 rcpt.from = m.to
106
107 # pass original message (before sending receipt)
108 m.to = jid
109 m.from = "#{users_num}@#{ARGV[0]}"
110
111 puts 'XRESPONSE0: ' + m.inspect
112 write_to_stream m
113
114 # send a delivery receipt back to the sender
115 # TODO: send only when requested per XEP-0184
116 # TODO: pass receipts from target if supported
117
118 # TODO: put in member/instance variable
119 rcpt['id'] = SecureRandom.uuid
120 rcvd = Nokogiri::XML::Node.new 'received', rcpt.document
121 rcvd['xmlns'] = 'urn:xmpp:receipts'
122 rcvd['id'] = m.id
123 rcpt.add_child(rcvd)
124
125 puts 'XRESPONSE1: ' + rcpt.inspect
126 write_to_stream rcpt
127 end
128
129 def self.call_catapult(
130 token, secret, m, pth, body=nil,
131 head={}, code=[200], respond_with=:body
132 )
133 EM::HttpRequest.new(
134 "https://api.catapult.inetwork.com/#{pth}"
135 ).public_send(
136 m,
137 head: {
138 'Authorization' => [token, secret]
139 }.merge(head),
140 body: body
141 ).then { |http|
142 puts "API response to send: #{http.response} with code"\
143 " response.code #{http.response_header.status}"
144
145 if code.include?(http.response_header.status)
146 case respond_with
147 when :body
148 http.response
149 when :headers
150 http.response_header
151 else
152 http
153 end
154 else
155 EMPromise.reject(http.response_header.status)
156 end
157 }
158 end
159
160 def self.to_catapult(s, murl, num_dest, user_id, token, secret, usern)
161 extra = if murl
162 {
163 media: murl
164 }
165 else
166 {
167 receiptRequested: 'all',
168 callbackUrl: ARGV[4]
169 }
170 end
171
172 call_catapult(
173 token,
174 secret,
175 :post,
176 "v1/users/#{user_id}/messages",
177 JSON.dump(extra.merge(
178 from: usern,
179 to: num_dest,
180 text: s.respond_to?(:body) ? s.body : '',
181 tag:
182 # callbacks need id and resourcepart
183 WEBrick::HTTPUtils.escape(s.id.to_s) +
184 ' ' +
185 WEBrick::HTTPUtils.escape(
186 s.from.resource.to_s
187 )
188 )),
189 {'Content-Type' => 'application/json'},
190 [201]
191 ).catch {
192 # TODO: add text; mention code number
193 EMPromise.reject(
194 [:cancel, 'internal-server-error']
195 )
196 }
197 end
198
199 def self.validate_num(num)
200 EMPromise.resolve(num.to_s).then { |num_dest|
201 if num_dest =~ /\A\+?[0-9]+(?:;.*)?\Z/
202 next num_dest if num_dest[0] == '+'
203 shortcode = extract_shortcode(num_dest)
204 next shortcode if shortcode
205 end
206 # TODO: text re num not (yet) supportd/implmentd
207 EMPromise.reject([:cancel, 'item-not-found'])
208 }
209 end
210
211 def self.fetch_catapult_cred_for(jid)
212 cred_key = "catapult_cred-#{jid.stripped}"
213 REDIS.lrange(cred_key, 0, 3).then { |creds|
214 if creds.length < 4
215 # TODO: add text re credentials not registered
216 EMPromise.reject(
217 [:auth, 'registration-required']
218 )
219 else
220 creds
221 end
222 }
223 end
224
225 message :chat?, :body do |m|
226 EMPromise.all([
227 validate_num(m.to.node),
228 fetch_catapult_cred_for(m.from)
229 ]).then { |(num_dest, creds)|
230 jid_key = "catapult_jid-#{num_dest}"
231 REDIS.get(jid_key).then { |jid|
232 [jid, num_dest] + creds
233 }
234 }.then { |(jid, num_dest, *creds)|
235 # if destination user is in the system pass on directly
236 if jid
237 pass_on_message(m, creds.last, jid)
238 else
239 to_catapult(m, nil, num_dest, *creds)
240 end
241 }.catch { |e|
242 if e.is_a?(Array) && e.length == 2
243 write_to_stream error_msg(m.reply, m.body, *e)
244 else
245 EMPromise.reject(e)
246 end
247 }
248 end
249
250 def self.user_cap_identities
251 [{category: 'client', type: 'sms'}]
252 end
253
254 def self.user_cap_features
255 [
256 "urn:xmpp:receipts",
257 "urn:xmpp:jingle:1", "urn:xmpp:jingle:transports:ibb:1",
258
259 # TODO: add more efficient file transfer mechanisms
260 #"urn:xmpp:jingle:transports:s5b:1",
261
262 # TODO: MUST add all reasonable vers of file-transfer
263 #"urn:xmpp:jingle:apps:file-transfer:4"
264 "urn:xmpp:jingle:apps:file-transfer:3"
265 ]
266 end
267
268 presence :subscribe? do |p|
269 puts "PRESENCE1: #{p.inspect}"
270
271 # subscriptions are allowed from anyone - send reply immediately
272 msg = Blather::Stanza::Presence.new
273 msg.to = p.from
274 msg.from = p.to
275 msg.type = :subscribed
276
277 puts 'RESPONSE5a: ' + msg.inspect
278 write_to_stream msg
279
280 # send a <presence> immediately; not automatically probed for it
281 # TODO: refactor so no "presence :probe? do |p|" duplicate below
282 caps = Blather::Stanza::Capabilities.new
283 # TODO: user a better node URI (?)
284 caps.node = 'http://catapult.sgx.soprani.ca/'
285 caps.identities = user_cap_identities
286 caps.features = user_cap_features
287
288 msg = caps.c
289 msg.to = p.from
290 msg.from = p.to.to_s + '/sgx'
291
292 puts 'RESPONSE5b: ' + msg.inspect
293 write_to_stream msg
294
295 # need to subscribe back so Conversations displays images inline
296 msg = Blather::Stanza::Presence.new
297 msg.to = p.from.to_s.split('/', 2)[0]
298 msg.from = p.to.to_s.split('/', 2)[0]
299 msg.type = :subscribe
300
301 puts 'RESPONSE5c: ' + msg.inspect
302 write_to_stream msg
303 end
304
305 presence :probe? do |p|
306 puts 'PRESENCE2: ' + p.inspect
307
308 caps = Blather::Stanza::Capabilities.new
309 # TODO: user a better node URI (?)
310 caps.node = 'http://catapult.sgx.soprani.ca/'
311 caps.identities = user_cap_identities
312 caps.features = user_cap_features
313
314 msg = caps.c
315 msg.to = p.from
316 msg.from = p.to.to_s + '/sgx'
317
318 puts 'RESPONSE6: ' + msg.inspect
319 write_to_stream msg
320 end
321
322 iq '/iq/ns:jingle', ns: 'urn:xmpp:jingle:1' do |i, jn|
323 puts "IQj: #{i.inspect}"
324
325 if jn[0]['action'] == 'transport-accept'
326 puts "REPLY0: #{i.reply.inspect}"
327 write_to_stream i.reply
328 next
329 elsif jn[0]['action'] == 'session-terminate'
330 # TODO: unexpected (usually we do this; handle?)
331 puts "TERMINATED"
332 next
333 elsif jn[0]['action'] == 'transport-info'
334 # TODO: unexpected, but should handle in a nice way
335 puts "FAIL!!!"
336 next
337 elsif i.type == :error
338 # TODO: do something, maybe terminating the connection
339 puts 'ERROR!!!'
340 next
341 end
342
343 # TODO: should probably confirm we got session-initiate here
344
345 write_to_stream i.reply
346 puts "RESPONSE8: #{i.reply.inspect}"
347
348 msg = Blather::Stanza::Iq.new :set
349 msg.to = i.from
350 msg.from = i.to
351
352 cn = jn.children.find { |v| v.element_name == "content" }
353 puts 'CN-name: ' + cn['name']
354 puts 'JN-sid: ' + jn[0]['sid']
355
356 ibb_found = false
357 last_sid = ''
358 cn.children.each do |child|
359 if child.element_name == 'transport'
360 puts 'TPORT: ' + child.namespace.href
361 last_sid = child['sid']
362 if 'urn:xmpp:jingle:transports:ibb:1' ==
363 child.namespace.href
364
365 ibb_found = true
366 break
367 end
368 end
369 end
370
371 j = Nokogiri::XML::Node.new 'jingle', msg.document
372 j['xmlns'] = 'urn:xmpp:jingle:1'
373 j['sid'] = jn[0]['sid']
374 msg.add_child(j)
375
376 content = Nokogiri::XML::Node.new 'content', msg.document
377 content['name'] = cn['name']
378 content['creator'] = 'initiator'
379 j.add_child(content)
380
381 transport = Nokogiri::XML::Node.new 'transport', msg.document
382 # TODO: make block-size more variable and/or dependent on sender
383 transport['block-size'] = '4096'
384 transport['xmlns'] = 'urn:xmpp:jingle:transports:ibb:1'
385 if ibb_found
386 transport['sid'] = last_sid
387 j['action'] = 'session-accept'
388 j['responder'] = i.from
389
390 dsc = Nokogiri::XML::Node.new 'description', msg.document
391 dsc['xmlns'] = 'urn:xmpp:jingle:apps:file-transfer:3'
392 content.add_child(dsc)
393 else
394 # for Conversations - it tries s5b even if caps ibb-only
395 transport['sid'] = SecureRandom.uuid
396 j['action'] = 'transport-replace'
397 j['initiator'] = i.from
398 end
399 content.add_child(transport)
400
401 @jingle_sids[transport['sid']] = jn[0]['sid']
402
403 # TODO: save <date> as well? Gajim sends, Conversations does not
404 # TODO: save/validate <size> with eventual full received length
405 fname =
406 cn
407 .children.find { |v| v.element_name == "description" }
408 .children.find { |w| w.element_name == "offer" }
409 .children.find { |x| x.element_name == "file" }
410 .children.find { |y| y.element_name == "name" }
411 @jingle_fnames[transport['sid']] = fname.text
412
413 puts "RESPONSE9: #{msg.inspect}"
414 write_to_stream msg
415 end
416
417 iq '/iq/ns:open', ns: 'http://jabber.org/protocol/ibb' do |i, on|
418 puts "IQo: #{i.inspect}"
419
420 @partial_data[on[0]['sid']] = ''
421 write_to_stream i.reply
422 end
423
424 iq '/iq/ns:data', ns: 'http://jabber.org/protocol/ibb' do |i, dn|
425 @partial_data[dn[0]['sid']] += Base64.decode64(dn[0].text)
426 write_to_stream i.reply
427 end
428
429 iq '/iq/ns:close', ns: 'http://jabber.org/protocol/ibb' do |i, cn|
430 puts "IQc: #{i.inspect}"
431 write_to_stream i.reply
432
433 EMPromise.all([
434 validate_num(i.to.node),
435 fetch_catapult_cred_for(i.from)
436 ]).then { |(num_dest, creds)|
437 # Gajim bug: <close/> has Jingle (not transport) sid; fix later
438 if not @jingle_fnames.key? cn[0]['sid']
439 puts 'ERROR: Not found in filename map: ' + cn[0]['sid']
440
441 next EMPromise.reject(:done)
442 # TODO: in case only Gajim's <data/> bug fixed, add map:
443 #cn[0]['sid'] = @jingle_tsids[cn[0]['sid']]
444 end
445
446 # upload cached data to server (before success reply)
447 media_name =
448 "#{Time.now.utc.iso8601}_#{SecureRandom.uuid}"\
449 "_#{@jingle_fnames[cn[0]['sid']]}"
450 puts 'name to save: ' + media_name
451
452 path = "/v1/users/#{creds.first}/media/#{media_name}"
453
454 EMPromise.all([
455 call_catapult(
456 *creds[1..2],
457 :put,
458 path,
459 @partial_data[cn[0]['sid']]
460 ).catch {
461 EMPromise.reject([
462 :cancel, 'internal-server-error'
463 ])
464 },
465 to_catapult(
466 i,
467 "https://api.catapult.inetwork.com/" +
468 path,
469 num_dest,
470 *creds
471 )
472 ])
473 }.then {
474 @partial_data[cn[0]['sid']] = ''
475
476 # received the complete file so now close the stream
477 msg = Blather::Stanza::Iq.new :set
478 msg.to = i.from
479 msg.from = i.to
480
481 j = Nokogiri::XML::Node.new 'jingle', msg.document
482 j['xmlns'] = 'urn:xmpp:jingle:1'
483 j['action'] = 'session-terminate'
484 j['sid'] = @jingle_sids[cn[0]['sid']]
485 msg.add_child(j)
486
487 r = Nokogiri::XML::Node.new 'reason', msg.document
488 s = Nokogiri::XML::Node.new 'success', msg.document
489 r.add_child(s)
490 j.add_child(r)
491
492 puts 'RESPONSE1: ' + msg.inspect
493 write_to_stream msg
494 }.catch { |e|
495 if e.is_a?(Array) && e.length == 2
496 write_to_stream error_msg(i.reply, nil, *e)
497 elsif e != :done
498 EMPromise.reject(e)
499 end
500 }
501 end
502
503 iq '/iq/ns:query', ns: 'http://jabber.org/protocol/disco#items' do |i|
504 write_to_stream i.reply
505 end
506
507 iq '/iq/ns:query', ns: 'http://jabber.org/protocol/disco#info' do |i|
508 # respond to capabilities request for an sgx-catapult number JID
509 if i.to.node
510 # TODO: confirm the node URL is expected using below
511 #puts "XR[node]: #{xpath_result[0]['node']}"
512
513 msg = i.reply
514 msg.identities = user_cap_identities
515 msg.features = user_cap_features
516
517 puts 'RESPONSE7: ' + msg.inspect
518 write_to_stream msg
519 next
520 end
521
522 # respond to capabilities request for sgx-catapult itself
523 msg = i.reply
524 msg.identities = [{
525 name: 'Soprani.ca Gateway to XMPP - Catapult',
526 type: 'sms', category: 'gateway'
527 }]
528 msg.features = [
529 "jabber:iq:register",
530 "jabber:iq:gateway",
531 "jabber:iq:private",
532 "http://jabber.org/protocol/disco#info",
533 "http://jabber.org/protocol/commands",
534 "http://jabber.org/protocol/muc"
535 ]
536 write_to_stream msg
537 end
538
539 def self.check_then_register(i, *creds)
540 jid_key = "catapult_jid-#{creds.last}"
541 bare_jid = i.from.stripped
542 cred_key = "catapult_cred-#{bare_jid}"
543
544 REDIS.get(jid_key).then { |existing_jid|
545 if existing_jid && existing_jid != bare_jid
546 # TODO: add/log text: credentials exist already
547 EMPromise.reject([:cancel, 'conflict'])
548 end
549 }.then {
550 REDIS.lrange(cred_key, 0, 3)
551 }.then { |existing_creds|
552 # TODO: add/log text: credentials exist already
553 if existing_creds.length == 4 && creds != existing_creds
554 EMPromise.reject([:cancel, 'conflict'])
555 elsif existing_creds.length < 4
556 REDIS.rpush(cred_key, *creds).then { |length|
557 if length != 4
558 EMPromise.reject([
559 :cancel,
560 'internal-server-error'
561 ])
562 end
563 }
564 end
565 }.then {
566 # not necessary if existing_jid non-nil, easier this way
567 REDIS.set(jid_key, bare_jid)
568 }.then { |result|
569 if result != 'OK'
570 # TODO: add txt re push failure
571 EMPromise.reject(
572 [:cancel, 'internal-server-error']
573 )
574 end
575 }.then {
576 write_to_stream i.reply
577 }
578 end
579
580 def self.creds_from_registration_query(qn)
581 xn = qn.children.find { |v| v.element_name == "x" }
582
583 if xn
584 xn.children.each_with_object({}) do |field, h|
585 next if field.element_name != "field"
586 val = field.children.find { |v|
587 v.element_name == "value"
588 }
589
590 case field['var']
591 when 'nick'
592 h[:user_id] = val.text
593 when 'username'
594 h[:api_token] = val.text
595 when 'password'
596 h[:api_secret] = val.text
597 when 'phone'
598 h[:phone_num] = val.text
599 else
600 # TODO: error
601 puts "?: #{field['var']}"
602 end
603 end
604 else
605 qn.children.each_with_object({}) do |field, h|
606 case field.element_name
607 when "nick"
608 h[:user_id] = field.text
609 when "username"
610 h[:api_token] = field.text
611 when "password"
612 h[:api_secret] = field.text
613 when "phone"
614 h[:phone_num] = field.text
615 end
616 end
617 end.values_at(:user_id, :api_token, :api_secret, :phone_num)
618 end
619
620 def self.process_registration(i, qn)
621 EMPromise.resolve(
622 qn.children.find { |v| v.element_name == "remove" }
623 ).then { |rn|
624 if rn
625 puts "received <remove/> - ignoring for now..."
626 EMPromise.reject(:done)
627 else
628 creds_from_registration_query(qn)
629 end
630 }.then { |user_id, api_token, api_secret, phone_num|
631 if phone_num[0] == '+'
632 [user_id, api_token, api_secret, phone_num]
633 else
634 # TODO: add text re number not (yet) supported
635 EMPromise.reject([:cancel, 'item-not-found'])
636 end
637 }.then { |user_id, api_token, api_secret, phone_num|
638 call_catapult(
639 api_token,
640 api_secret,
641 :get,
642 "v1/users/#{user_id}/phoneNumbers/#{phone_num}"
643 ).then { |response|
644 params = JSON.parse(response)
645 if params['numberState'] == 'enabled'
646 check_then_register(
647 i,
648 user_id,
649 api_token,
650 api_secret,
651 phone_num
652 )
653 else
654 # TODO: add text re number disabled
655 EMPromise.reject([:modify, 'not-acceptable'])
656 end
657 }
658 }.catch { |e|
659 EMPromise.reject(case e
660 when 401
661 # TODO: add text re bad credentials
662 [:auth, 'not-authorized']
663 when 404
664 # TODO: add text re number not found or disabled
665 [:cancel, 'item-not-found']
666 when Integer
667 [:modify, 'not-acceptable']
668 else
669 e
670 end)
671 }
672 end
673
674 def self.registration_form(orig, existing_number=nil)
675 msg = Nokogiri::XML::Node.new 'query', orig.document
676 msg['xmlns'] = 'jabber:iq:register'
677
678 if existing_number
679 msg.add_child(
680 Nokogiri::XML::Node.new(
681 'registered', msg.document
682 )
683 )
684 end
685
686 n1 = Nokogiri::XML::Node.new(
687 'instructions', msg.document
688 )
689 n1.content = "Enter the information from your Account "\
690 "page as well as the Phone Number\nin your "\
691 "account you want to use (ie. '+12345678901')"\
692 ".\nUser Id is nick, API Token is username, "\
693 "API Secret is password, Phone Number is phone"\
694 ".\n\nThe source code for this gateway is at "\
695 "https://gitlab.com/ossguy/sgx-catapult ."\
696 "\nCopyright (C) 2017 Denver Gingerich and "\
697 "others, licensed under AGPLv3+."
698 n2 = Nokogiri::XML::Node.new 'nick', msg.document
699 n3 = Nokogiri::XML::Node.new 'username', msg.document
700 n4 = Nokogiri::XML::Node.new 'password', msg.document
701 n5 = Nokogiri::XML::Node.new 'phone', msg.document
702 n5.content = existing_number.to_s
703 msg.add_child(n1)
704 msg.add_child(n2)
705 msg.add_child(n3)
706 msg.add_child(n4)
707 msg.add_child(n5)
708
709 x = Blather::Stanza::X.new :form, [
710 {
711 required: true, type: :"text-single",
712 label: 'User Id', var: 'nick'
713 },
714 {
715 required: true, type: :"text-single",
716 label: 'API Token', var: 'username'
717 },
718 {
719 required: true, type: :"text-private",
720 label: 'API Secret', var: 'password'
721 },
722 {
723 required: true, type: :"text-single",
724 label: 'Phone Number', var: 'phone',
725 value: existing_number.to_s
726 }
727 ]
728 x.title = 'Register for '\
729 'Soprani.ca Gateway to XMPP - Catapult'
730 x.instructions = "Enter the details from your Account "\
731 "page as well as the Phone Number\nin your "\
732 "account you want to use (ie. '+12345678901')"\
733 ".\n\nThe source code for this gateway is at "\
734 "https://gitlab.com/ossguy/sgx-catapult ."\
735 "\nCopyright (C) 2017 Denver Gingerich and "\
736 "others, licensed under AGPLv3+."
737 msg.add_child(x)
738
739 orig.add_child(msg)
740
741 return orig
742 end
743
744 iq '/iq/ns:query', ns: 'jabber:iq:register' do |i, qn|
745 puts "IQ: #{i.inspect}"
746
747 case i.type
748 when :set
749 process_registration(i, qn)
750 when :get
751 bare_jid = i.from.stripped
752 cred_key = "catapult_cred-#{bare_jid}"
753 REDIS.lindex(cred_key, 3).then { |existing_number|
754 reply = registration_form(i.reply, existing_number)
755 puts "RESPONSE2: #{reply.inspect}"
756 write_to_stream reply
757 }
758 else
759 # Unknown IQ, ignore for now
760 EMPromise.reject(:done)
761 end.catch { |e|
762 if e.is_a?(Array) && e.length == 2
763 write_to_stream error_msg(i.reply, qn, *e)
764 elsif e != :done
765 EMPromise.reject(e)
766 end
767 }.catch(&method(:panic))
768 end
769
770 subscription(:request?) do |s|
771 # TODO: are these the best to return? really need '!' here?
772 #write_to_stream s.approve!
773 #write_to_stream s.request!
774 end
775
776 iq :get? do |i|
777 write_to_stream error_msg(i.reply, i.children, 'cancel', 'feature-not-implemented')
778 end
779
780 iq :set? do |i|
781 write_to_stream error_msg(i.reply, i.children, 'cancel', 'feature-not-implemented')
782 end
783end
784
785class ReceiptMessage < Blather::Stanza
786 def self.new(to=nil)
787 node = super :message
788 node.to = to
789 node
790 end
791end
792
793class WebhookHandler < Goliath::API
794 use Goliath::Rack::Params
795
796 def send_media(from, to, media_url)
797 # we assume media_url is of the form (always the case so far):
798 # https://api.catapult.inetwork.com/v1/users/[uid]/media/[file]
799
800 # the caller must guarantee that 'to' is a bare JID
801 proxy_url = ARGV[6] + to + '/' + media_url.split('/', 8)[7]
802
803 puts 'ORIG_URL: ' + media_url
804 puts 'PROX_URL: ' + proxy_url
805
806 # put URL in the body (so Conversations will still see it)...
807 msg = Blather::Stanza::Message.new(to, proxy_url)
808 msg.from = from
809
810 # ...but also provide URL in XEP-0066 (OOB) fashion
811 # TODO: confirm client supports OOB or don't send this
812 x = Nokogiri::XML::Node.new 'x', msg.document
813 x['xmlns'] = 'jabber:x:oob'
814
815 urln = Nokogiri::XML::Node.new 'url', msg.document
816 urlc = Nokogiri::XML::Text.new proxy_url, msg.document
817
818 urln.add_child(urlc)
819 x.add_child(urln)
820 msg.add_child(x)
821
822 SGXcatapult.write(msg)
823
824 rescue Exception => e
825 puts 'Shutting down gateway due to exception 012: ' + e.message
826 SGXcatapult.shutdown
827 puts 'Gateway has terminated.'
828 EM.stop
829 end
830
831 def response(env)
832 puts 'ENV: ' + env.to_s
833
834 users_num = ''
835 others_num = ''
836 if params['direction'] == 'in'
837 users_num = params['to']
838 others_num = params['from']
839 elsif params['direction'] == 'out'
840 users_num = params['from']
841 others_num = params['to']
842 else
843 # TODO: exception or similar
844 puts "big problem: '" + params['direction'] + "'" + body
845 return [200, {}, "OK"]
846 end
847
848 puts 'BODY - messageId: ' + params['messageId'] +
849 ', eventType: ' + params['eventType'] +
850 ', time: ' + params['time'] +
851 ', direction: ' + params['direction'] +
852 ', state: ' + params['state'] +
853 ', deliveryState: ' + (params['deliveryState'] ?
854 params['deliveryState'] : 'NONE') +
855 ', deliveryCode: ' + (params['deliveryCode'] ?
856 params['deliveryCode'] : 'NONE') +
857 ', deliveryDesc: ' + (params['deliveryDescription'] ?
858 params['deliveryDescription'] : 'NONE') +
859 ', tag: ' + (params['tag'] ? params['tag'] : 'NONE') +
860 ', media: ' + (params['media'] ? params['media'].to_s :
861 'NONE')
862
863 if others_num[0] != '+'
864 # TODO: check that others_num actually a shortcode first
865 others_num +=
866 ';phone-context=ca-us.phone-context.soprani.ca'
867 end
868
869 jid_key = "catapult_jid-#{users_num}"
870 bare_jid = REDIS.get(jid_key).promise.sync
871
872 if !bare_jid
873 puts "jid_key (#{jid_key}) DNE; Catapult misconfigured?"
874
875 # TODO: likely not appropriate; give error to Catapult?
876 # TODO: add text re credentials not being registered
877 #write_to_stream error_msg(m.reply, m.body, :auth,
878 # 'registration-required')
879 return [200, {}, "OK"]
880 end
881
882 msg = ''
883 case params['direction']
884 when 'in'
885 text = ''
886 case params['eventType']
887 when 'sms'
888 text = params['text']
889 when 'mms'
890 has_media = false
891 params['media'].each do |media_url|
892 if not media_url.end_with?(
893 '.smil', '.txt', '.xml'
894 )
895
896 has_media = true
897 send_media(
898 others_num + '@' +
899 ARGV[0],
900 bare_jid, media_url
901 )
902 end
903 end
904
905 if params['text'].empty?
906 if not has_media
907 text = '[suspected group msg '\
908 'with no text (odd)]'
909 end
910 else
911 text = if has_media
912 # TODO: write/use a caption XEP
913 params['text']
914 else
915 '[suspected group msg '\
916 '(recipient list not '\
917 'available) with '\
918 'following text] ' +
919 params['text']
920 end
921 end
922
923 # ie. if text param non-empty or had no media
924 if not text.empty?
925 msg = Blather::Stanza::Message.new(
926 bare_jid, text)
927 msg.from = others_num + '@' + ARGV[0]
928 SGXcatapult.write(msg)
929 end
930
931 return [200, {}, "OK"]
932 else
933 text = "unknown type (#{params['eventType']})"\
934 " with text: " + params['text']
935
936 # TODO: log/notify of this properly
937 puts text
938 end
939
940 msg = Blather::Stanza::Message.new(bare_jid, text)
941 else # per prior switch, this is: params['direction'] == 'out'
942 tag_parts = params['tag'].split(/ /, 2)
943 id = WEBrick::HTTPUtils.unescape(tag_parts[0])
944 resourcepart = WEBrick::HTTPUtils.unescape(tag_parts[1])
945
946 case params['deliveryState']
947 when 'not-delivered'
948 # create a bare message like the one user sent
949 msg = Blather::Stanza::Message.new(
950 others_num + '@' + ARGV[0])
951 msg.from = bare_jid + '/' + resourcepart
952 msg['id'] = id
953
954 # create an error reply to the bare message
955 msg = Blather::StanzaError.new(
956 msg,
957 'recipient-unavailable',
958 :wait
959 ).to_node
960 when 'delivered'
961 msg = ReceiptMessage.new(bare_jid)
962
963 # TODO: put in member/instance variable
964 msg['id'] = SecureRandom.uuid
965
966 # TODO: send only when requested per XEP-0184
967 rcvd = Nokogiri::XML::Node.new(
968 'received',
969 msg.document
970 )
971 rcvd['xmlns'] = 'urn:xmpp:receipts'
972 rcvd['id'] = id
973 msg.add_child(rcvd)
974 when 'waiting'
975 # can't really do anything with it; nice to know
976 puts "message with id #{id} waiting"
977 return [200, {}, "OK"]
978 else
979 # TODO: notify somehow of unknown state receivd?
980 puts "message with id #{id} has "\
981 "other state #{params['deliveryState']}"
982 return [200, {}, "OK"]
983 end
984
985 puts "RESPONSE4: #{msg.inspect}"
986 end
987
988 msg.from = others_num + '@' + ARGV[0]
989 SGXcatapult.write(msg)
990
991 [200, {}, "OK"]
992
993 rescue Exception => e
994 puts 'Shutting down gateway due to exception 013: ' + e.message
995 SGXcatapult.shutdown
996 puts 'Gateway has terminated.'
997 EM.stop
998 end
999end
1000
1001at_exit do
1002 $stdout.sync = true
1003
1004 puts "Soprani.ca/SMS Gateway for XMPP - Catapult\n"\
1005 "==>> last commit of this version is " + `git rev-parse HEAD` + "\n"
1006
1007 if ARGV.size != 7
1008 puts "Usage: sgx-catapult.rb <component_jid> "\
1009 "<component_password> <server_hostname> "\
1010 "<server_port> <delivery_receipt_url> "\
1011 "<http_listen_port> <mms_proxy_prefix_url>"
1012 exit 0
1013 end
1014
1015 t = Time.now
1016 puts "LOG %d.%09d: starting...\n\n" % [t.to_i, t.nsec]
1017
1018 EM.run do
1019 REDIS = EM::Hiredis.connect
1020
1021 SGXcatapult.run
1022
1023 # required when using Prosody otherwise disconnects on 6-hour inactivity
1024 EM.add_periodic_timer(3600) do
1025 msg = Blather::Stanza::Iq::Ping.new(:get, 'localhost')
1026 msg.from = ARGV[0]
1027 SGXcatapult.write(msg)
1028 end
1029
1030 server = Goliath::Server.new('0.0.0.0', ARGV[5].to_i)
1031 server.api = WebhookHandler.new
1032 server.app = Goliath::Rack::Builder.build(server.api.class, server.api)
1033 server.logger = Log4r::Logger.new('goliath')
1034 server.logger.add(Log4r::StdoutOutputter.new('console'))
1035 server.logger.level = Log4r::INFO
1036 server.start do
1037 ["INT", "TERM"].each do |sig|
1038 trap(sig) do
1039 puts 'Shutting down gateway...'
1040 SGXcatapult.shutdown
1041
1042 puts 'Gateway has terminated.'
1043 EM.stop
1044 end
1045 end
1046 end
1047 end
1048end