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 return [400, {}, "Missing params\n"] if jparams['to'].empty?
1015
1016 puts "BODY - messageId: #{jparams['id']}" \
1017 ", eventType: #{type}" \
1018 ", time: #{jparams['time']}" \
1019 ", direction: #{jparams['direction']}" \
1020 ", deliveryState: #{jparams['deliveryState'] || 'NONE'}" \
1021 ", errorCode: #{jparams['errorCode'] || 'NONE'}" \
1022 ", description: #{jparams['description'] || 'NONE'}" \
1023 ", tag: #{jparams['tag'] || 'NONE'}" \
1024 ", media: #{jparams['media'] || 'NONE'}"
1025
1026 if others_num[0] != '+'
1027 # TODO: check that others_num actually a shortcode first
1028 others_num +=
1029 ';phone-context=ca-us.phone-context.soprani.ca'
1030 end
1031
1032 bare_jid = @registration_repo.find_jid(users_num).sync
1033
1034 if !bare_jid
1035 puts "jid_key for (#{users_num}) DNE; BW API misconfigured?"
1036
1037 return [403, {}, "Customer not found\n"]
1038 end
1039
1040 msg = nil
1041 case jparams['direction']
1042 when 'in'
1043 text = ''
1044 case type
1045 when 'message-received'
1046 # TODO: handle group chat, and fix above
1047 text = jparams['text']
1048
1049 if jparams['to'].length > 1
1050 msg = Blather::Stanza::Message.new(
1051 Blather::JID.new(bare_jid).domain
1052 )
1053 msg.body = text unless text&.empty?
1054
1055 addrs = Nokogiri::XML::Node.new(
1056 'addresses', msg.document)
1057 addrs['xmlns'] = 'http://jabber.org/' \
1058 'protocol/address'
1059
1060 addr1 = Nokogiri::XML::Node.new(
1061 'address', msg.document)
1062 addr1['type'] = 'to'
1063 addr1['jid'] = bare_jid
1064 addrs.add_child(addr1)
1065
1066 jparams['to'].reject(
1067 # Don't send to the same person twice,
1068 # and don't send to the person who sent it
1069 &[users_num, others_num].method(:include?)
1070 ).each do |receiver|
1071 addrn = Nokogiri::XML::Node.new(
1072 'address', msg.document)
1073 addrn['type'] = 'to'
1074 addrn['uri'] = "sms:#{receiver}"
1075 addrn['delivered'] = 'true'
1076 addrs.add_child(addrn)
1077 end
1078
1079 msg.add_child(addrs)
1080
1081 # TODO: delete
1082 puts "RESPONSE9: #{msg.inspect}"
1083 end
1084
1085 done = Array(jparams['media']).map { |media_url|
1086 unless media_url.end_with?(
1087 '.smil', '.txt', '.xml'
1088 )
1089 SGXbwmsgsv2.send_media(
1090 others_num + '@' +
1091 ARGV[0],
1092 bare_jid, media_url,
1093 nil, nil, msg
1094 )
1095 end
1096
1097 media_url
1098 }.empty?.then { |no_media|
1099 next if no_media
1100 next unless text&.empty?
1101 next unless jparams['to'].length > 1
1102
1103 next [200, {}, "OK"]
1104 }
1105
1106 return done unless done.nil?
1107 else
1108 text = "unknown type (#{type})"\
1109 " with text: " + jparams['text']
1110
1111 # TODO: log/notify of this properly
1112 puts text
1113 end
1114
1115 # If text is not empty, but there isn't a msg,
1116 # we need to construct a msg to convey that text
1117 unless msg || text&.empty?
1118 msg = Blather::Stanza::Message.new(
1119 bare_jid,
1120 # TODO: The numbers, what do they mean?
1121 text.gsub(/[\u0000-\u0008\u000b-\u001f]/, "")
1122 )
1123 msg.document.encoding = "utf-8"
1124 msg.chat_state = nil
1125 end
1126 else # per prior switch, this is: jparams['direction'] == 'out'
1127 tag_parts = jparams['tag'].split(/ /, 2)
1128 id = WEBrick::HTTPUtils.unescape(tag_parts[0])
1129 resourcepart = WEBrick::HTTPUtils.unescape(tag_parts[1])
1130
1131 # TODO: remove this hack
1132 if jparams['to'].length > 1
1133 puts "WARN! group no rcpt: #{users_num}"
1134 return [200, {}, "OK"]
1135 end
1136
1137 case type
1138 when 'message-failed'
1139 # create a bare message like the one user sent
1140 msg = Blather::Stanza::Message.new(
1141 others_num + '@' + ARGV[0])
1142 msg.from = bare_jid + '/' + resourcepart
1143 msg['id'] = id
1144
1145 # TODO: add 'errorCode' and/or 'description' val
1146 # create an error reply to the bare message
1147 msg = msg.as_error(
1148 'recipient-unavailable',
1149 :wait,
1150 params['_json'][0]['description']
1151 )
1152
1153 when 'message-delivered'
1154
1155 msg = ReceiptMessage.new(bare_jid)
1156
1157 # TODO: put in member/instance variable
1158 msg['id'] = SecureRandom.uuid
1159
1160 # TODO: send only when requested per XEP-0184
1161 rcvd = Nokogiri::XML::Node.new(
1162 'received',
1163 msg.document
1164 )
1165 rcvd['xmlns'] = 'urn:xmpp:receipts'
1166 rcvd['id'] = id
1167 msg.add_child(rcvd)
1168
1169 # TODO: make prettier: this should be done above
1170 others_num = params['_json'][0]['to']
1171 else
1172 # TODO: notify somehow of unknown state receivd?
1173 puts "message with id #{id} has "\
1174 "other type #{type}"
1175 return [200, {}, "OK"]
1176 end
1177
1178 puts "RESPONSE4: #{msg.inspect}"
1179 end
1180
1181 # if message-failed, we already set msg.from
1182 # moreover, we said `msg = msg.as_error`, and StanzaError
1183 msg.from = others_num + '@' + ARGV[0] if msg.respond_to?(:from=)
1184 SGXbwmsgsv2.write(msg)
1185
1186 # Emit event to messages stream
1187 case [jparams['direction'], type]
1188 when ['in', 'message-received']
1189 if !env['HTTP_X_JMP_RESEND_OF'].to_s.empty?
1190 MessageEvent::ResendIn.new(
1191 original_stream_id: env['HTTP_X_JMP_RESEND_OF'],
1192 original_bandwidth_id: jparams['id'],
1193 owner: jparams['owner']
1194 ).emit(REDIS)
1195 else
1196 media_urls = Array(jparams['media']).reject { |u|
1197 u.end_with?('.smil', '.txt', '.xml')
1198 }
1199 MessageEvent::In.new(
1200 timestamp: jparams['time'],
1201 from: jparams['from'],
1202 to: jparams['to'],
1203 owner: jparams['owner'],
1204 bandwidth_id: jparams['id'],
1205 body: jparams['text'].to_s,
1206 media_urls: media_urls
1207 ).emit(REDIS)
1208 end
1209 when ['out', 'message-failed']
1210 tag_parts = jparams['tag'].split(/ /, 2)
1211 stanza_id = WEBrick::HTTPUtils.unescape(tag_parts[0])
1212 MessageEvent::Failed.new(
1213 timestamp: jparams['time'],
1214 stanza_id: stanza_id,
1215 bandwidth_id: jparams['id'],
1216 error_code: jparams['errorCode'].to_s,
1217 error_description: jparams['description'].to_s
1218 ).emit(REDIS)
1219 when ['out', 'message-delivered']
1220 tag_parts = jparams['tag'].split(/ /, 2)
1221 stanza_id = WEBrick::HTTPUtils.unescape(tag_parts[0])
1222 MessageEvent::Delivered.new(
1223 timestamp: jparams['time'],
1224 stanza_id: stanza_id,
1225 bandwidth_id: jparams['id']
1226 ).emit(REDIS)
1227 end
1228
1229 [200, {}, "OK"]
1230 rescue Exception => e
1231 Sentry.capture_exception(e)
1232 puts 'Shutting down gateway due to exception 013: ' + e.message
1233 SGXbwmsgsv2.shutdown
1234 puts 'Gateway has terminated.'
1235 EM.stop unless ENV['ENV'] == 'test'
1236 end
1237end
1238
1239at_exit do
1240 $stdout.sync = true
1241
1242 puts "Soprani.ca/SMS Gateway for XMPP - Bandwidth API V2\n"\
1243 "==>> last commit of this version is " + `git rev-parse HEAD` + "\n"
1244
1245 if ARGV.size != 7
1246 puts "Usage: sgx-bwmsgsv2.rb <component_jid> "\
1247 "<component_password> <server_hostname> "\
1248 "<server_port> <application_id> "\
1249 "<http_listen_port> <mms_proxy_prefix_url>"
1250 exit 0
1251 end
1252
1253 t = Time.now
1254 puts "LOG %d.%09d: starting...\n\n" % [t.to_i, t.nsec]
1255
1256 EM.run do
1257 REDIS = EM::Hiredis.connect
1258
1259 SGXbwmsgsv2.run
1260
1261 # required when using Prosody otherwise disconnects on 6-hour inactivity
1262 EM.add_periodic_timer(3600) do
1263 msg = Blather::Stanza::Iq::Ping.new(:get, 'localhost')
1264 msg.from = ARGV[0]
1265 SGXbwmsgsv2.write(msg)
1266 end
1267
1268 server = Goliath::Server.new('0.0.0.0', ARGV[5].to_i)
1269 server.api = WebhookHandler.new
1270 server.app = Goliath::Rack::Builder.build(server.api.class, server.api)
1271 server.logger = Log4r::Logger.new('goliath')
1272 server.logger.add(Log4r::StdoutOutputter.new('console'))
1273 server.logger.level = Log4r::INFO
1274 server.start do
1275 ["INT", "TERM"].each do |sig|
1276 trap(sig) do
1277 EM.defer do
1278 puts 'Shutting down gateway...'
1279 SGXbwmsgsv2.shutdown
1280
1281 puts 'Gateway has terminated.'
1282 EM.stop
1283 end
1284 end
1285 end
1286 end
1287 end
1288end unless ENV['ENV'] == 'test'