1#!/usr/bin/env ruby
2# frozen_string_literal: true
3
4# Copyright (C) 2017-2020 Denver Gingerich <denver@ossguy.com>
5# Copyright (C) 2017 Stephen Paul Weber <singpolyma@singpolyma.net>
6#
7# This file is part of sgx-bwmsgsv2.
8#
9# sgx-bwmsgsv2 is free software: you can redistribute it and/or modify it under
10# the terms of the GNU Affero General Public License as published by the Free
11# Software Foundation, either version 3 of the License, or (at your option) any
12# later version.
13#
14# sgx-bwmsgsv2 is distributed in the hope that it will be useful, but WITHOUT
15# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
16# FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more
17# details.
18#
19# You should have received a copy of the GNU Affero General Public License along
20# with sgx-bwmsgsv2. If not, see <http://www.gnu.org/licenses/>.
21
22require 'blather/client/dsl'
23require 'em-hiredis'
24require 'em-http-request'
25require 'json'
26require 'multibases'
27require 'multihashes'
28require 'securerandom'
29require "sentry-ruby"
30require 'time'
31require 'uri'
32require 'webrick'
33
34require 'goliath/api'
35require 'goliath/server'
36require 'log4r'
37
38require 'em_promise'
39require 'em-synchrony'
40
41require_relative 'lib/bandwidth_error'
42require_relative 'lib/bandwidth_tn_options'
43require_relative 'lib/message_event'
44require_relative 'lib/registration_repo'
45
46Sentry.init
47
48# List of supported MIME types from Bandwidth - https://support.bandwidth.com/hc/en-us/articles/360014128994-What-MMS-file-types-are-supported-
49MMS_MIME_TYPES = [
50 "application/json",
51 "application/ogg",
52 "application/pdf",
53 "application/rtf",
54 "application/zip",
55 "application/x-tar",
56 "application/xml",
57 "application/gzip",
58 "application/x-bzip2",
59 "application/x-gzip",
60 "application/smil",
61 "application/javascript",
62 "audio/mp4",
63 "audio/mpeg",
64 "audio/ogg",
65 "audio/flac",
66 "audio/webm",
67 "audio/wav",
68 "audio/amr",
69 "audio/3gpp",
70 "image/bmp",
71 "image/gif",
72 "image/jpeg",
73 "image/pjpeg",
74 "image/png",
75 "image/svg+xml",
76 "image/tiff",
77 "image/webp",
78 "image/x-icon",
79 "text/css",
80 "text/csv",
81 "text/calendar",
82 "text/plain",
83 "text/javascript",
84 "text/vcard",
85 "text/vnd.wap.wml",
86 "text/xml",
87 "video/avi",
88 "video/mp4",
89 "video/mpeg",
90 "video/ogg",
91 "video/quicktime",
92 "video/webm",
93 "video/x-ms-wmv",
94 "video/x-flv"
95]
96
97def panic(e)
98 Sentry.capture_exception(e)
99 puts "Shutting down gateway due to exception: #{e.message}"
100 puts e.backtrace
101 SGXbwmsgsv2.shutdown
102 puts 'Gateway has terminated.'
103 EM.stop
104end
105
106EM.error_handler(&method(:panic))
107
108def extract_shortcode(dest)
109 num, context = dest.split(';', 2)
110 num if context == 'phone-context=ca-us.phone-context.soprani.ca'
111end
112
113def anonymous_tel?(dest)
114 dest.split(';', 2)[1] == 'phone-context=anonymous.phone-context.soprani.ca'
115end
116
117class SGXClient < Blather::Client
118 def register_handler(type, *guards, &block)
119 super(type, *guards) { |*args| wrap_handler(*args, &block) }
120 end
121
122 def register_handler_before(type, *guards, &block)
123 check_handler(type, guards)
124 handler = lambda { |*args| wrap_handler(*args, &block) }
125
126 @handlers[type] ||= []
127 @handlers[type].unshift([guards, handler])
128 end
129
130protected
131
132 def wrap_handler(*args)
133 v = yield(*args)
134 v = v.sync if ENV['ENV'] == 'test' && v.is_a?(Promise)
135 v.catch(&method(:panic)) if v.is_a?(Promise)
136 true # Do not run other handlers unless throw :pass
137 rescue Exception => e
138 panic(e)
139 end
140end
141
142# TODO: keep in sync with jmp-acct_bot.rb, and eventually put in common location
143module CatapultSettingFlagBits
144 VOICEMAIL_TRANSCRIPTION_DISABLED = 0
145 MMS_ON_OOB_URL = 1
146end
147
148module SGXbwmsgsv2
149 extend Blather::DSL
150
151 @registration_repo = RegistrationRepo.new
152 @client = SGXClient.new
153 @gateway_features = [
154 "http://jabber.org/protocol/disco#info",
155 "http://jabber.org/protocol/address/",
156 "jabber:iq:register",
157 "http://jabber.org/protocol/commands"
158 ]
159
160 def self.run
161 # TODO: read/save ARGV[7] creds to local variables
162 client.run
163 end
164
165 # so classes outside this module can write messages, too
166 def self.write(stanza)
167 client.write(stanza)
168 end
169
170 def self.before_handler(type, *guards, &block)
171 client.register_handler_before(type, *guards, &block)
172 end
173
174 def self.send_media(from, to, media_url, desc=nil, subject=nil, m=nil)
175 # we assume media_url is one of these (always the case so far):
176 # https://messaging.bandwidth.com/api/v2/users/[u]/media/[path]
177
178 puts 'ORIG_URL: ' + media_url
179 usr = to
180 if media_url.start_with?('https://messaging.bandwidth.com/api/v2/users/')
181 pth = media_url.split('/', 9)[8]
182 # the caller must guarantee that 'to' is a bare JID
183 media_url = ARGV[6] + WEBrick::HTTPUtils.escape(usr) + '/' + pth
184 puts 'PROX_URL: ' + media_url
185 end
186
187 msg = m ? m.copy : Blather::Stanza::Message.new(to, "")
188 msg.from = from
189 msg.subject = subject if subject
190
191 # provide URL in XEP-0066 (OOB) fashion
192 x = Nokogiri::XML::Node.new 'x', msg.document
193 x['xmlns'] = 'jabber:x:oob'
194
195 urln = Nokogiri::XML::Node.new 'url', msg.document
196 urlc = Nokogiri::XML::Text.new media_url, msg.document
197 urln.add_child(urlc)
198 x.add_child(urln)
199
200 if desc
201 descn = Nokogiri::XML::Node.new('desc', msg.document)
202 descc = Nokogiri::XML::Text.new(desc, msg.document)
203 descn.add_child(descc)
204 x.add_child(descn)
205 end
206
207 msg.add_child(x)
208
209 write(msg)
210 rescue Exception => e
211 panic(e)
212 end
213
214 setup ARGV[0], ARGV[1], ARGV[2], ARGV[3], nil, nil, async: true
215
216 def self.pass_on_message(m, users_num, jid)
217 # Capture destination before modifying m.to
218 dest_num = m.to.node
219
220 # setup delivery receipt; similar to a reply
221 rcpt = ReceiptMessage.new(m.from.stripped)
222 rcpt.from = m.to
223
224 # pass original message (before sending receipt)
225 m.to = jid
226 m.from = "#{users_num}@#{ARGV[0]}"
227
228 puts 'XRESPONSE0: ' + m.inspect
229 write_to_stream m
230
231 # Emit pass-through event
232 oob_url = m.at("oob|x > oob|url", oob: "jabber:x:oob")&.text
233 MessageEvent::Thru.new(
234 timestamp: Time.now,
235 from: users_num,
236 to: [dest_num],
237 message_id: m.id,
238 body: m.body.to_s,
239 media_urls: [oob_url].compact
240 ).emit(REDIS)
241
242 # send a delivery receipt back to the sender
243 # TODO: send only when requested per XEP-0184
244 # TODO: pass receipts from target if supported
245
246 # TODO: put in member/instance variable
247 rcpt['id'] = SecureRandom.uuid
248 rcvd = Nokogiri::XML::Node.new 'received', rcpt.document
249 rcvd['xmlns'] = 'urn:xmpp:receipts'
250 rcvd['id'] = m.id
251 rcpt.add_child(rcvd)
252
253 puts 'XRESPONSE1: ' + rcpt.inspect
254 write_to_stream rcpt
255 end
256
257 def self.call_catapult(
258 token, secret, m, pth, body=nil,
259 head={}, code=[200], respond_with=:body
260 )
261 # pth looks like one of:
262 # "api/v2/users/#{user_id}/[endpoint_name]"
263
264 url_prefix = ''
265
266 # TODO: need to make a separate thing for voice.bw.c eventually
267 if pth.start_with? 'api/v2/users'
268 url_prefix = 'https://messaging.bandwidth.com/'
269 end
270
271 EM::HttpRequest.new(
272 url_prefix + pth
273 ).public_send(
274 m,
275 head: {
276 'Authorization' => [token, secret]
277 }.merge(head),
278 body: body
279 ).then { |http|
280 puts "API response to send: #{http.response} with code"\
281 " response.code #{http.response_header.status}"
282
283 if code.include?(http.response_header.status)
284 case respond_with
285 when :body
286 http.response
287 when :headers
288 http.response_header
289 else
290 http
291 end
292 else
293 EMPromise.reject(
294 BandwidthError.for(http.response_header.status, http.response)
295 )
296 end
297 }
298 end
299
300 def self.to_catapult_possible_oob(s, num_dest, user_id, token, secret,
301 usern)
302 un = s.at("oob|x > oob|url", oob: "jabber:x:oob")
303 unless un
304 puts "MMSOOB: no url node found so process as normal"
305 return to_catapult(s, nil, num_dest, user_id, token,
306 secret, usern)
307 end
308 puts "MMSOOB: found a url node - checking if to make MMS..."
309
310 body = s.respond_to?(:body) ? s.body : ''
311 EM::HttpRequest.new(un.text, tls: { verify_peer: true }).head.then { |http|
312 # If content is too large, or MIME type is not supported, place the link inside the body and do not send MMS.
313 if http.response_header["CONTENT_LENGTH"].to_i > 3500000 ||
314 !MMS_MIME_TYPES.include?(http.response_header["CONTENT_TYPE"])
315 unless body.include?(un.text)
316 s.body = body.empty? ? un.text : "#{body}\n#{un.text}"
317 end
318 to_catapult(s, nil, num_dest, user_id, token, secret, usern)
319 else # If size is less than ~3.5MB, strip the link from the body and attach media in the body.
320 # some clients send URI in both body & <url/> so delete
321 s.body = body.sub(/\s*#{Regexp.escape(un.text)}\s*$/, '')
322
323 puts "MMSOOB: url text is '#{un.text}'"
324 puts "MMSOOB: the body is '#{body.to_s.strip}'"
325
326 puts "MMSOOB: sending MMS since found OOB & user asked"
327 to_catapult(s, un.text, num_dest, user_id, token, secret, usern)
328 end
329 }
330 end
331
332 def self.to_catapult(s, murl, num_dest, user_id, token, secret, usern)
333 body = s.respond_to?(:body) ? s.body.to_s : ''
334 if murl.to_s.empty? && body.strip.empty?
335 return EMPromise.reject(
336 [:modify, 'policy-violation']
337 )
338 end
339
340 segment_size = body.ascii_only? ? 160 : 70
341 if !murl && ENV["MMS_PATH"] && body.length > segment_size*3
342 file = Multibases.pack(
343 'base58btc',
344 Multihashes.encode(Digest::SHA256.digest(body), "sha2-256")
345 ).to_s
346 File.open("#{ENV['MMS_PATH']}/#{file}", "w") { |fh| fh.write body }
347 murl = "#{ENV['MMS_URL']}/#{file}.txt"
348 body = ""
349 end
350
351 extra = {}
352 extra[:media] = murl if murl
353
354 call_catapult(
355 token,
356 secret,
357 :post,
358 "api/v2/users/#{user_id}/messages",
359 JSON.dump(extra.merge(
360 from: usern,
361 to: num_dest,
362 text: body,
363 applicationId: ARGV[4],
364 tag:
365 # callbacks need id and resourcepart
366 WEBrick::HTTPUtils.escape(s.id.to_s) +
367 ' ' +
368 WEBrick::HTTPUtils.escape(
369 s.from.resource.to_s
370 )
371 )),
372 {'Content-Type' => 'application/json'},
373 [201]
374 ).then { |response|
375 parsed = JSON.parse(response) rescue {}
376 MessageEvent::Out.new(
377 timestamp: Time.now,
378 from: usern,
379 to: Array(num_dest),
380 message_id: parsed["id"] || s.id,
381 body: body,
382 media_urls: [murl].compact
383 ).emit(REDIS)
384 response
385 }.catch { |e|
386 EMPromise.reject(
387 [:cancel, 'internal-server-error', e.message]
388 )
389 }
390 end
391
392 def self.validate_num(m)
393 # if sent to SGX domain use https://wiki.soprani.ca/SGX/GroupMMS
394 if m.to == ARGV[0]
395 an = m.children.find { |v| v.element_name == "addresses" }
396 if not an
397 return EMPromise.reject(
398 [:cancel, 'item-not-found']
399 )
400 end
401 puts "ADRXEP: found an addresses node - iterate addrs.."
402
403 nums = []
404 an.children.each do |e|
405 num = ''
406 type = ''
407 e.attributes.each do |c|
408 if c[0] == 'type'
409 if c[1] != 'to'
410 # TODO: error
411 end
412 type = c[1].to_s
413 elsif c[0] == 'uri'
414 if !c[1].to_s.start_with? 'sms:'
415 # TODO: error
416 end
417 num = c[1].to_s[4..-1]
418 # TODO: confirm num validates
419 # TODO: else, error - unexpected name
420 end
421 end
422 if num.empty? or type.empty?
423 # TODO: error
424 end
425 nums << num
426 end
427 return nums
428 end
429
430 # if not sent to SGX domain, then assume destination is in 'to'
431 EMPromise.resolve(m.to.node.to_s).then { |num_dest|
432 if num_dest =~ /\A\+?[0-9]+(?:;.*)?\Z/
433 next num_dest if num_dest[0] == '+'
434
435 shortcode = extract_shortcode(num_dest)
436 next shortcode if shortcode
437 end
438
439 if anonymous_tel?(num_dest)
440 EMPromise.reject([:cancel, 'gone'])
441 else
442 # TODO: text re num not (yet) supportd/implmentd
443 EMPromise.reject([:cancel, 'item-not-found'])
444 end
445 }
446 end
447
448 def self.fetch_catapult_cred_for(jid)
449 @registration_repo.find(jid).then { |creds|
450 if creds.length < 4
451 # TODO: add text re credentials not registered
452 EMPromise.reject(
453 [:auth, 'registration-required']
454 )
455 else
456 creds
457 end
458 }
459 end
460
461 message :error? do |m|
462 # TODO: report it somewhere/somehow - eat for now so no err loop
463 puts "EATERROR1: #{m.inspect}"
464 end
465
466 message :body do |m|
467 EMPromise.all([
468 validate_num(m),
469 fetch_catapult_cred_for(m.from)
470 ]).then { |(num_dest, creds)|
471 @registration_repo.find_jid(num_dest).then { |jid|
472 [jid, num_dest] + creds
473 }
474 }.then { |(jid, num_dest, *creds)|
475 if jid
476 @registration_repo.find(jid).then { |other_user|
477 [jid, num_dest] + creds + [other_user.first]
478 }
479 else
480 [jid, num_dest] + creds + [nil]
481 end
482 }.then { |(jid, num_dest, *creds, other_user)|
483 # if destination user is in the system pass on directly
484 if other_user and not other_user.start_with? 'u-'
485 pass_on_message(m, creds.last, jid)
486 else
487 to_catapult_possible_oob(m, num_dest, *creds)
488 end
489 }.catch { |e|
490 if e.is_a?(Array) && (e.length == 2 || e.length == 3)
491 write_to_stream m.as_error(e[1], e[0], e[2])
492 else
493 EMPromise.reject(e)
494 end
495 }
496 end
497
498 def self.user_cap_identities
499 [{category: 'client', type: 'sms'}]
500 end
501
502 # TODO: must re-add stuff so can do ad-hoc commands
503 def self.user_cap_features
504 ["urn:xmpp:receipts"]
505 end
506
507 def self.add_gateway_feature(feature)
508 @gateway_features << feature
509 @gateway_features.uniq!
510 end
511
512 subscription :request? do |p|
513 puts "PRESENCE1: #{p.inspect}"
514
515 # subscriptions are allowed from anyone - send reply immediately
516 msg = Blather::Stanza::Presence.new
517 msg.to = p.from
518 msg.from = p.to
519 msg.type = :subscribed
520
521 puts 'RESPONSE5a: ' + msg.inspect
522 write_to_stream msg
523
524 # send a <presence> immediately; not automatically probed for it
525 # TODO: refactor so no "presence :probe? do |p|" duplicate below
526 caps = Blather::Stanza::Capabilities.new
527 # TODO: user a better node URI (?)
528 caps.node = 'http://catapult.sgx.soprani.ca/'
529 caps.identities = user_cap_identities
530 caps.features = user_cap_features
531
532 msg = caps.c
533 msg.to = p.from
534 msg.from = p.to.to_s + '/sgx'
535
536 puts 'RESPONSE5b: ' + msg.inspect
537 write_to_stream msg
538
539 # need to subscribe back so Conversations displays images inline
540 msg = Blather::Stanza::Presence.new
541 msg.to = p.from.to_s.split('/', 2)[0]
542 msg.from = p.to.to_s.split('/', 2)[0]
543 msg.type = :subscribe
544
545 puts 'RESPONSE5c: ' + msg.inspect
546 write_to_stream msg
547 end
548
549 presence :probe? do |p|
550 puts 'PRESENCE2: ' + p.inspect
551
552 caps = Blather::Stanza::Capabilities.new
553 # TODO: user a better node URI (?)
554 caps.node = 'http://catapult.sgx.soprani.ca/'
555 caps.identities = user_cap_identities
556 caps.features = user_cap_features
557
558 msg = caps.c
559 msg.to = p.from
560 msg.from = p.to.to_s + '/sgx'
561
562 puts 'RESPONSE6: ' + msg.inspect
563 write_to_stream msg
564 end
565
566 disco_items(
567 to: Blather::JID.new(ARGV[0]),
568 node: "http://jabber.org/protocol/commands"
569 ) do |i|
570 fetch_catapult_cred_for(i.from).then { |creds|
571 BandwidthTNOptions.tn_eligible_for_port_out_pin?(creds).then { |eligible|
572 reply = i.reply
573 reply.node = 'http://jabber.org/protocol/commands'
574
575 if eligible
576 reply.items = [
577 Blather::Stanza::DiscoItems::Item.new(
578 i.to,
579 'set-port-out-pin',
580 'Set Port-Out PIN'
581 )
582 ]
583 else
584 reply.items = []
585 end
586
587 puts 'RESPONSE_CMD_DISCO: ' + reply.inspect
588 write_to_stream reply
589 }
590 }.catch { |e|
591 if e.is_a?(Array) && [2, 3].include?(e.length)
592 write_to_stream i.as_error(e[1], e[0], e[2])
593 else
594 EMPromise.reject(e)
595 end
596 }
597 end
598
599 iq '/iq/ns:query', ns: 'http://jabber.org/protocol/disco#info' do |i|
600 # TODO: return error if i.type is :set - if it is :reply or
601 # :error it should be ignored (as the below does currently);
602 # review specification to see how to handle other type values
603 if i.type != :get
604 puts 'DISCO iq rcvd, of non-get type "' + i.type.to_s +
605 '" for message "' + i.inspect + '"; ignoring...'
606 next
607 end
608
609 # respond to capabilities request for an sgx-bwmsgsv2 number JID
610 if i.to.node
611 # TODO: confirm the node URL is expected using below
612 #puts "XR[node]: #{xpath_result[0]['node']}"
613
614 msg = i.reply
615 msg.node = i.node
616 msg.identities = user_cap_identities
617 msg.features = user_cap_features
618
619 puts 'RESPONSE7: ' + msg.inspect
620 write_to_stream msg
621 next
622 end
623
624 # respond to capabilities request for sgx-bwmsgsv2 itself
625 msg = i.reply
626 msg.node = i.node
627 msg.identities = [{
628 name: 'Soprani.ca Gateway to XMPP - Bandwidth API V2',
629 type: 'sms', category: 'gateway'
630 }]
631 msg.features = @gateway_features
632 write_to_stream msg
633 end
634
635 def self.check_then_register(i, *creds)
636 @registration_repo
637 .put(i.from, *creds)
638 .catch_only(RegistrationRepo::Conflict) { |e|
639 EMPromise.reject([:cancel, 'conflict', e.message])
640 }.then {
641 write_to_stream i.reply
642 }
643 end
644
645 def self.creds_from_registration_query(i)
646 if i.query.find_first("./ns:x", ns: "jabber:x:data")
647 [
648 i.form.field("nick")&.value,
649 i.form.field("username")&.value,
650 i.form.field("password")&.value,
651 i.form.field("phone")&.value
652 ]
653 else
654 [i.nick, i.username, i.password, i.phone]
655 end
656 end
657
658 def self.process_registration(i)
659 EMPromise.resolve(nil).then {
660 if i.remove?
661 @registration_repo.delete(i.from).then do
662 write_to_stream i.reply
663 EMPromise.reject(:done)
664 end
665 else
666 creds_from_registration_query(i)
667 end
668 }.then { |user_id, api_token, api_secret, phone_num|
669 if phone_num && phone_num[0] == '+'
670 [user_id, api_token, api_secret, phone_num]
671 else
672 # TODO: add text re number not (yet) supported
673 EMPromise.reject([:cancel, 'item-not-found'])
674 end
675 }.then { |user_id, api_token, api_secret, phone_num|
676 # TODO: find way to verify #{phone_num}, too
677 call_catapult(
678 api_token,
679 api_secret,
680 :get,
681 "api/v2/users/#{user_id}/media"
682 ).then { |response|
683 JSON.parse(response)
684 # TODO: confirm response is array - could be empty
685
686 puts "register got str #{response.to_s[0..999]}"
687
688 check_then_register(
689 i,
690 user_id,
691 api_token,
692 api_secret,
693 phone_num
694 )
695 }
696 }.catch_only(BandwidthError) { |e|
697 EMPromise.reject(case e.code
698 when 401
699 # TODO: add text re bad credentials
700 [:auth, 'not-authorized']
701 when 404
702 # TODO: add text re number not found or disabled
703 [:cancel, 'item-not-found']
704 else
705 [:modify, 'not-acceptable']
706 end)
707 }
708 end
709
710 def self.registration_form(orig, existing_number=nil)
711 orig.registered = !!existing_number
712
713 # TODO: update "User Id" x2 below (to "accountId"?), and others?
714 orig.instructions = "Enter the information from your Account "\
715 "page as well as the Phone Number\nin your "\
716 "account you want to use (ie. '+12345678901')"\
717 ".\nUser Id is nick, API Token is username, "\
718 "API Secret is password, Phone Number is phone"\
719 ".\n\nThe source code for this gateway is at "\
720 "https://gitlab.com/soprani.ca/sgx-bwmsgsv2 ."\
721 "\nCopyright (C) 2017-2020 Denver Gingerich "\
722 "and others, licensed under AGPLv3+."
723 orig.nick = ""
724 orig.username = ""
725 orig.password = ""
726 orig.phone = existing_number.to_s
727
728 orig.form.fields = [
729 {
730 required: true, type: :"text-single",
731 label: 'User Id', var: 'nick'
732 },
733 {
734 required: true, type: :"text-single",
735 label: 'API Token', var: 'username'
736 },
737 {
738 required: true, type: :"text-private",
739 label: 'API Secret', var: 'password'
740 },
741 {
742 required: true, type: :"text-single",
743 label: 'Phone Number', var: 'phone',
744 value: existing_number.to_s
745 }
746 ]
747 orig.form.title = 'Register for '\
748 'Soprani.ca Gateway to XMPP - Bandwidth API V2'
749 orig.form.instructions = "Enter the details from your Account "\
750 "page as well as the Phone Number\nin your "\
751 "account you want to use (ie. '+12345678901')"\
752 ".\n\nThe source code for this gateway is at "\
753 "https://gitlab.com/soprani.ca/sgx-bwmsgsv2 ."\
754 "\nCopyright (C) 2017-2020 Denver Gingerich "\
755 "and others, licensed under AGPLv3+."
756
757 orig
758 end
759
760 ibr do |i|
761 puts "IQ: #{i.inspect}"
762
763 case i.type
764 when :set
765 process_registration(i)
766 when :get
767 bare_jid = i.from.stripped
768 @registration_repo.find(bare_jid).then { |creds|
769 reply = registration_form(i.reply, creds.last)
770 puts "RESPONSE2: #{reply.inspect}"
771 write_to_stream reply
772 }
773 else
774 # Unknown IQ, ignore for now
775 EMPromise.reject(:done)
776 end.catch { |e|
777 if e.is_a?(Array) && (e.length == 2 || e.length == 3)
778 write_to_stream i.as_error(e[1], e[0], e[2])
779 elsif e != :done
780 EMPromise.reject(e)
781 end
782 }.catch(&method(:panic))
783 end
784
785 command :execute?, node: "set-port-out-pin", sessionid: nil do |iq|
786 # Ensure user is registered, but discard their credentials
787 # because we're just showing them a form.
788 fetch_catapult_cred_for(iq.from).then { |_creds|
789 reply = iq.reply
790 reply.node = 'set-port-out-pin'
791 reply.sessionid = SecureRandom.uuid
792 reply.status = :executing
793
794 form = Blather::Stanza::X.find_or_create(reply.command)
795 form.type = "form"
796 form.fields = [
797 {
798 var: 'pin',
799 type: 'text-private',
800 label: 'Port-Out PIN',
801 required: true
802 },
803 {
804 var: 'confirm_pin',
805 type: 'text-private',
806 label: 'Confirm PIN',
807 required: true
808 }
809 ]
810
811 reply.command.add_child(form)
812 reply.allowed_actions = [:complete]
813
814 puts "RESPONSE_CMD_FORM: #{reply.inspect}"
815 write_to_stream reply
816 }.catch { |e|
817 if e.is_a?(Array) && [2, 3].include?(e.length)
818 write_to_stream iq.as_error(e[1], e[0], e[2])
819 else
820 EMPromise.reject(e)
821 end
822 }.catch(&method(:panic))
823 end
824
825 command :complete?, node: "set-port-out-pin", sessionid: /./ do |iq|
826 pin = iq.form.field('pin')&.value
827 confirm_pin = iq.form.field('confirm_pin')&.value
828
829 if pin.nil? || confirm_pin.nil?
830 write_to_stream iq.as_error(
831 'bad-request',
832 :modify,
833 'PIN fields are required'
834 )
835 next
836 end
837
838 if pin != confirm_pin
839 write_to_stream iq.as_error(
840 'bad-request',
841 :modify,
842 'PIN confirmation does not match'
843 )
844 next
845 end
846
847 if pin !~ /\A[a-zA-Z0-9]{4,10}\z/
848 write_to_stream iq.as_error(
849 'bad-request',
850 :modify,
851 'PIN must be 4-10 alphanumeric characters'
852 )
853 next
854 end
855
856 fetch_catapult_cred_for(iq.from).then { |creds|
857 BandwidthTNOptions.set_port_out_pin(creds, pin).then {
858 reply = iq.reply
859 reply.node = 'set-port-out-pin'
860 reply.sessionid = iq.sessionid
861 reply.status = :completed
862 reply.note_type = :info
863 reply.note_text = 'Port-out PIN has been set successfully.'
864
865 write_to_stream reply
866 }.catch { |e|
867 reply = iq.reply
868 reply.node = 'set-port-out-pin'
869 reply.sessionid = iq.sessionid
870 reply.status = :completed
871 reply.note_type = :error
872 error_msg = if e.respond_to?(:message) && e.message.include?('not valid')
873 "Invalid phone number format. "\
874 "Please check your registered phone number."
875 elsif e.respond_to?(:message) && e.message.include?('ErrorCode')
876 "Bandwidth API error: #{e.message}"
877 else
878 "Failed to set port-out PIN. Please try again later."
879 end
880 reply.note_text = error_msg
881
882 write_to_stream reply
883 }
884 }.catch { |e|
885 if e.is_a?(Array) && [2, 3].include?(e.length)
886 write_to_stream iq.as_error(e[1], e[0], e[2])
887 else
888 EMPromise.reject(e)
889 end
890 }.catch(&method(:panic))
891 end
892
893 iq type: [:get, :set] do |iq|
894 write_to_stream(Blather::StanzaError.new(
895 iq,
896 'feature-not-implemented',
897 :cancel
898 ))
899 end
900end
901
902class ReceiptMessage < Blather::Stanza
903 def self.new(to=nil)
904 node = super :message
905 node.to = to
906 node
907 end
908end
909
910class WebhookHandler < Goliath::API
911 use Sentry::Rack::CaptureExceptions
912 use Goliath::Rack::Params
913
914 def response(env)
915 @registration_repo = RegistrationRepo.new
916 # TODO: add timestamp grab here, and MUST include ./tai version
917
918 puts 'ENV: ' + env.reject { |k| k == 'params' }.to_s
919
920 if params.empty?
921 puts 'PARAMS empty!'
922 return [200, {}, "OK"]
923 end
924
925 if env['REQUEST_URI'] != '/'
926 puts 'BADREQUEST1: non-/ request "' +
927 env['REQUEST_URI'] + '", method "' +
928 env['REQUEST_METHOD'] + '"'
929 return [200, {}, "OK"]
930 end
931
932 if env['REQUEST_METHOD'] != 'POST'
933 puts 'BADREQUEST2: non-POST request; URI: "' +
934 env['REQUEST_URI'] + '", method "' +
935 env['REQUEST_METHOD'] + '"'
936 return [200, {}, "OK"]
937 end
938
939 # TODO: process each message in list, not just first one
940 jparams = params.dig('_json', 0, 'message')
941 type = params.dig('_json', 0, 'type')
942
943 return [400, {}, "Missing params\n"] unless jparams && type
944
945 users_num, others_num = if jparams['direction'] == 'in'
946 [jparams['owner'], jparams['from']]
947 elsif jparams['direction'] == 'out'
948 [jparams['from'], jparams['owner']]
949 else
950 puts "big prob: '#{jparams['direction']}'"
951 return [400, {}, "OK"]
952 end
953
954 jparams['to'].reject! { |num|
955 num == users_num || num == others_num
956 }
957
958 return [400, {}, "Missing params\n"] unless users_num && others_num
959 return [400, {}, "Missing params\n"] unless jparams['to'].is_a?(Array)
960
961 puts "BODY - messageId: #{jparams['id']}" \
962 ", eventType: #{type}" \
963 ", time: #{jparams['time']}" \
964 ", direction: #{jparams['direction']}" \
965 ", deliveryState: #{jparams['deliveryState'] || 'NONE'}" \
966 ", errorCode: #{jparams['errorCode'] || 'NONE'}" \
967 ", description: #{jparams['description'] || 'NONE'}" \
968 ", tag: #{jparams['tag'] || 'NONE'}" \
969 ", media: #{jparams['media'] || 'NONE'}"
970
971 if others_num[0] != '+'
972 # TODO: check that others_num actually a shortcode first
973 others_num +=
974 ';phone-context=ca-us.phone-context.soprani.ca'
975 end
976
977 bare_jid = @registration_repo.find_jid(users_num).sync
978
979 if !bare_jid
980 puts "jid_key for (#{users_num}) DNE; BW API misconfigured?"
981
982 return [403, {}, "Customer not found\n"]
983 end
984
985 msg = nil
986 case jparams['direction']
987 when 'in'
988 text = ''
989 case type
990 when 'sms'
991 text = jparams['text']
992 when 'mms'
993 has_media = false
994
995 if jparams['text'].empty?
996 if not has_media
997 text = '[suspected group msg '\
998 'with no text (odd)]'
999 end
1000 else
1001 text = if has_media
1002 # TODO: write/use a caption XEP
1003 jparams['text']
1004 else
1005 '[suspected group msg '\
1006 '(recipient list not '\
1007 'available) with '\
1008 'following text] ' +
1009 jparams['text']
1010 end
1011 end
1012
1013 # ie. if text param non-empty or had no media
1014 if not text.empty?
1015 msg = Blather::Stanza::Message.new(
1016 bare_jid, text)
1017 msg.from = others_num + '@' + ARGV[0]
1018 SGXbwmsgsv2.write(msg)
1019 end
1020
1021 return [200, {}, "OK"]
1022 when 'message-received'
1023 # TODO: handle group chat, and fix above
1024 text = jparams['text']
1025
1026 if jparams['to'].length > 1
1027 msg = Blather::Stanza::Message.new(
1028 Blather::JID.new(bare_jid).domain,
1029 text
1030 )
1031
1032 addrs = Nokogiri::XML::Node.new(
1033 'addresses', msg.document)
1034 addrs['xmlns'] = 'http://jabber.org/' \
1035 'protocol/address'
1036
1037 addr1 = Nokogiri::XML::Node.new(
1038 'address', msg.document)
1039 addr1['type'] = 'to'
1040 addr1['jid'] = bare_jid
1041 addrs.add_child(addr1)
1042
1043 jparams['to'].each do |receiver|
1044 addrn = Nokogiri::XML::Node.new(
1045 'address', msg.document)
1046 addrn['type'] = 'to'
1047 addrn['uri'] = "sms:#{receiver}"
1048 addrn['delivered'] = 'true'
1049 addrs.add_child(addrn)
1050 end
1051
1052 msg.add_child(addrs)
1053
1054 # TODO: delete
1055 puts "RESPONSE9: #{msg.inspect}"
1056 end
1057
1058 Array(jparams['media']).each do |media_url|
1059 unless media_url.end_with?(
1060 '.smil', '.txt', '.xml'
1061 )
1062 has_media = true
1063 SGXbwmsgsv2.send_media(
1064 others_num + '@' +
1065 ARGV[0],
1066 bare_jid, media_url,
1067 nil, nil, msg
1068 )
1069 end
1070 end
1071 else
1072 text = "unknown type (#{type})"\
1073 " with text: " + jparams['text']
1074
1075 # TODO: log/notify of this properly
1076 puts text
1077 end
1078
1079 if not msg
1080 msg = Blather::Stanza::Message.new(bare_jid, text)
1081 end
1082 else # per prior switch, this is: jparams['direction'] == 'out'
1083 tag_parts = jparams['tag'].split(/ /, 2)
1084 id = WEBrick::HTTPUtils.unescape(tag_parts[0])
1085 resourcepart = WEBrick::HTTPUtils.unescape(tag_parts[1])
1086
1087 # TODO: remove this hack
1088 if jparams['to'].length > 1
1089 puts "WARN! group no rcpt: #{users_num}"
1090 return [200, {}, "OK"]
1091 end
1092
1093 case type
1094 when 'message-failed'
1095 # create a bare message like the one user sent
1096 msg = Blather::Stanza::Message.new(
1097 others_num + '@' + ARGV[0])
1098 msg.from = bare_jid + '/' + resourcepart
1099 msg['id'] = id
1100
1101 # TODO: add 'errorCode' and/or 'description' val
1102 # create an error reply to the bare message
1103 msg = msg.as_error(
1104 'recipient-unavailable',
1105 :wait,
1106 jparams['description']
1107 )
1108
1109 # TODO: make prettier: this should be done above
1110 others_num = params['_json'][0]['to']
1111 when 'message-delivered'
1112
1113 msg = ReceiptMessage.new(bare_jid)
1114
1115 # TODO: put in member/instance variable
1116 msg['id'] = SecureRandom.uuid
1117
1118 # TODO: send only when requested per XEP-0184
1119 rcvd = Nokogiri::XML::Node.new(
1120 'received',
1121 msg.document
1122 )
1123 rcvd['xmlns'] = 'urn:xmpp:receipts'
1124 rcvd['id'] = id
1125 msg.add_child(rcvd)
1126
1127 # TODO: make prettier: this should be done above
1128 others_num = params['_json'][0]['to']
1129 else
1130 # TODO: notify somehow of unknown state receivd?
1131 puts "message with id #{id} has "\
1132 "other type #{type}"
1133 return [200, {}, "OK"]
1134 end
1135
1136 puts "RESPONSE4: #{msg.inspect}"
1137 end
1138
1139 msg.from = others_num + '@' + ARGV[0]
1140 SGXbwmsgsv2.write(msg)
1141
1142 # Emit event to messages stream
1143 case [jparams['direction'], type]
1144 when ['in', 'message-received']
1145 media_urls = Array(jparams['media']).reject { |u|
1146 u.end_with?('.smil', '.txt', '.xml')
1147 }
1148 MessageEvent::In.new(
1149 timestamp: jparams['time'],
1150 from: jparams['from'],
1151 to: jparams['to'],
1152 message_id: jparams['id'],
1153 body: jparams['text'].to_s,
1154 media_urls: media_urls
1155 ).emit(REDIS)
1156 when ['out', 'message-failed']
1157 tag_parts = jparams['tag'].split(/ /, 2)
1158 msg_id = WEBrick::HTTPUtils.unescape(tag_parts[0])
1159 MessageEvent::Failed.new(
1160 timestamp: jparams['time'],
1161 message_id: msg_id,
1162 error_code: jparams['errorCode'].to_s,
1163 error_description: jparams['description'].to_s
1164 ).emit(REDIS)
1165 when ['out', 'message-delivered']
1166 tag_parts = jparams['tag'].split(/ /, 2)
1167 msg_id = WEBrick::HTTPUtils.unescape(tag_parts[0])
1168 MessageEvent::Delivered.new(
1169 timestamp: jparams['time'],
1170 message_id: msg_id
1171 ).emit(REDIS)
1172 end
1173
1174 [200, {}, "OK"]
1175 rescue Exception => e
1176 Sentry.capture_exception(e)
1177 puts 'Shutting down gateway due to exception 013: ' + e.message
1178 SGXbwmsgsv2.shutdown
1179 puts 'Gateway has terminated.'
1180 EM.stop
1181 end
1182end
1183
1184at_exit do
1185 $stdout.sync = true
1186
1187 puts "Soprani.ca/SMS Gateway for XMPP - Bandwidth API V2\n"\
1188 "==>> last commit of this version is " + `git rev-parse HEAD` + "\n"
1189
1190 if ARGV.size != 7
1191 puts "Usage: sgx-bwmsgsv2.rb <component_jid> "\
1192 "<component_password> <server_hostname> "\
1193 "<server_port> <application_id> "\
1194 "<http_listen_port> <mms_proxy_prefix_url>"
1195 exit 0
1196 end
1197
1198 t = Time.now
1199 puts "LOG %d.%09d: starting...\n\n" % [t.to_i, t.nsec]
1200
1201 EM.run do
1202 REDIS = EM::Hiredis.connect
1203
1204 SGXbwmsgsv2.run
1205
1206 # required when using Prosody otherwise disconnects on 6-hour inactivity
1207 EM.add_periodic_timer(3600) do
1208 msg = Blather::Stanza::Iq::Ping.new(:get, 'localhost')
1209 msg.from = ARGV[0]
1210 SGXbwmsgsv2.write(msg)
1211 end
1212
1213 server = Goliath::Server.new('0.0.0.0', ARGV[5].to_i)
1214 server.api = WebhookHandler.new
1215 server.app = Goliath::Rack::Builder.build(server.api.class, server.api)
1216 server.logger = Log4r::Logger.new('goliath')
1217 server.logger.add(Log4r::StdoutOutputter.new('console'))
1218 server.logger.level = Log4r::INFO
1219 server.start do
1220 ["INT", "TERM"].each do |sig|
1221 trap(sig) do
1222 EM.defer do
1223 puts 'Shutting down gateway...'
1224 SGXbwmsgsv2.shutdown
1225
1226 puts 'Gateway has terminated.'
1227 EM.stop
1228 end
1229 end
1230 end
1231 end
1232 end
1233end unless ENV['ENV'] == 'test'