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