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