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