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