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 }
454 end
455
456 def self.to_catapult(s, murl, num_dest, user_id, token, secret, usern)
457 body = s.respond_to?(:body) ? s.body.to_s : ''
458 if murl.to_s.empty? && body.strip.empty?
459 return EMPromise.reject(
460 [:modify, 'policy-violation']
461 )
462 end
463
464 if body.downcase.match?(BADWORDS)
465 return EMPromise.reject([
466 :wait,
467 'recipient-unavailable',
468 'Single message blocked by carrier content policy, see https://blog.jmp.chat/b/sms-censorship for details'
469 ])
470 end
471
472 if body =~ /\u2063/
473 return EMPromise.reject([
474 :wait,
475 'recipient-unavailable',
476 'Please contact JMP support about your message'
477 ])
478 end
479
480 if usern.match?(WHISPER_NUMBER)
481 return EMPromise.reject([
482 :cancel,
483 'recipient-unavailable',
484 'Please register with a backend'
485 ])
486 end
487
488 segment_size = body.ascii_only? ? 160 : 70
489 if !murl && ENV["MMS_PATH"] && num_dest =~ /^\+?1/ && body.length > segment_size*3
490 file = Multibases.pack(
491 'base58btc',
492 Multihashes.encode(Digest::SHA256.digest(body), "sha2-256")
493 ).to_s
494 File.open("#{ENV['MMS_PATH']}/#{file}", "w") { |fh| fh.write body }
495 murl = "#{ENV['MMS_URL']}/#{file}.txt"
496 body = ""
497 end
498
499 extra = {}
500 extra[:media] = murl if murl
501
502 call_catapult(
503 token,
504 secret,
505 :post,
506 "api/v2/users/#{user_id}/messages",
507 JSON.dump(extra.merge(
508 from: usern,
509 to: num_dest,
510 text: body,
511 applicationId: ARGV[4],
512 tag:
513 # callbacks need id and resourcepart
514 WEBrick::HTTPUtils.escape(s.id.to_s) +
515 ' ' +
516 WEBrick::HTTPUtils.escape(
517 s.from.resource.to_s
518 )
519 )),
520 {'Content-Type' => 'application/json'},
521 [201, 202]
522 ).then { |response|
523 parsed = JSON.parse(response) rescue {}
524 MessageEvent::Out.new(
525 timestamp: parsed["time"] || Time.now,
526 owner: usern,
527 from: usern,
528 to: Array(num_dest),
529 stanza_id: s.id.to_s,
530 bandwidth_id: parsed["id"],
531 body: body,
532 media_urls: [murl].compact
533 ).emit(REDIS)
534 response
535 }.catch { |e|
536 EMPromise.reject(
537 [:cancel, 'internal-server-error', e.message]
538 )
539 }
540 end
541
542 def self.validate_num(m)
543 # if sent to SGX domain use https://wiki.soprani.ca/SGX/GroupMMS
544 if m.to == ARGV[0]
545 an = m.children.find { |v| v.element_name == "addresses" }
546 if not an
547 return EMPromise.reject(
548 [:cancel, 'item-not-found']
549 )
550 end
551 LOG.debug "Found addresses node, iterating"
552
553 nums = []
554 an.children.each do |e|
555 num = ''
556 type = ''
557 e.attributes.each do |c|
558 if c[0] == 'type'
559 if c[1] != 'to'
560 # TODO: error
561 end
562 type = c[1].to_s
563 elsif c[0] == 'uri'
564 if !c[1].to_s.start_with? 'sms:'
565 # TODO: error
566 end
567 num = c[1].to_s[4..-1]
568 # TODO: confirm num validates
569 # TODO: else, error - unexpected name
570 end
571 end
572 if num.empty? or type.empty?
573 # TODO: error
574 end
575 nums << num
576 end
577 return nums
578 end
579
580 # if not sent to SGX domain, then assume destination is in 'to'
581 EMPromise.resolve(m.to.node.to_s).then { |num_dest|
582 if num_dest =~ /\A\+?[0-9]+(?:;.*)?\Z/
583 next num_dest if num_dest[0] == '+'
584
585 shortcode = extract_shortcode(num_dest)
586 next shortcode if shortcode
587 end
588
589 if anonymous_tel?(num_dest)
590 EMPromise.reject([:cancel, 'gone'])
591 else
592 # TODO: text re num not (yet) supportd/implmentd
593 EMPromise.reject([:cancel, 'item-not-found'])
594 end
595 }
596 end
597
598 def self.fetch_catapult_cred_for(jid)
599 @registration_repo.find(jid).then { |creds|
600 if creds.length < 4
601 # TODO: add text re credentials not registered
602 EMPromise.reject(
603 [:auth, 'registration-required']
604 )
605 else
606 creds
607 end
608 }
609 end
610
611 message :error? do |m|
612 # TODO: report it somewhere/somehow - eat for now so no err loop
613 LOG.warn("Eating error stanza", stanza: m.inspect)
614 true
615 end
616
617 message(->(m) { m.body || m.at("oob|x > oob|url", oob: "jabber:x:oob") }) do |m|
618 EMPromise.all([
619 validate_num(m),
620 fetch_catapult_cred_for(m.from)
621 ]).then { |(num_dest, creds)|
622 @registration_repo.find_jid(num_dest).then { |jid|
623 [jid, num_dest] + creds
624 }
625 }.then { |(jid, num_dest, *creds)|
626 if jid
627 @registration_repo.find(jid).then { |other_user|
628 [jid, num_dest] + creds + [other_user.first]
629 }
630 else
631 [jid, num_dest] + creds + [nil]
632 end
633 }.then { |(jid, num_dest, *creds, other_user)|
634 # if destination user is in the system pass on directly
635 if other_user and not other_user.start_with? 'u-'
636 pass_on_message(m, creds.last, jid)
637 else
638 to_catapult_possible_oob(m, num_dest, *creds)
639 end
640 }.catch { |e|
641 if e.is_a?(Array) && (e.length == 2 || e.length == 3)
642 write_to_stream m.as_error(e[1], e[0], e[2])
643 else
644 EMPromise.reject(e)
645 end
646 }
647 end
648
649 def self.user_cap_identities
650 [{category: 'client', type: 'sms'}]
651 end
652
653 # TODO: must re-add stuff so can do ad-hoc commands
654 def self.user_cap_features
655 ["urn:xmpp:receipts"]
656 end
657
658 def self.add_gateway_feature(feature)
659 @gateway_features << feature
660 @gateway_features.uniq!
661 end
662
663 subscription :request? do |p|
664 # subscriptions are allowed from anyone - send reply immediately
665 msg = Blather::Stanza::Presence.new
666 msg.to = p.from
667 msg.from = p.to
668 msg.type = :subscribed
669
670 write_to_stream msg
671
672 # send a <presence> immediately; not automatically probed for it
673 # TODO: refactor so no "presence :probe? do |p|" duplicate below
674 caps = Blather::Stanza::Capabilities.new
675 # TODO: user a better node URI (?)
676 caps.node = 'http://catapult.sgx.soprani.ca/'
677 caps.identities = user_cap_identities
678 caps.features = user_cap_features
679
680 msg = caps.c
681 msg.to = p.from
682 msg.from = p.to.to_s + '/sgx'
683
684 write_to_stream msg
685
686 # need to subscribe back so Conversations displays images inline
687 msg = Blather::Stanza::Presence.new
688 msg.to = p.from.to_s.split('/', 2)[0]
689 msg.from = p.to.to_s.split('/', 2)[0]
690 msg.type = :subscribe
691
692 write_to_stream msg
693
694 true
695 end
696
697 presence :probe? do |p|
698 caps = Blather::Stanza::Capabilities.new
699 # TODO: user a better node URI (?)
700 caps.node = 'http://catapult.sgx.soprani.ca/'
701 caps.identities = user_cap_identities
702 caps.features = user_cap_features
703
704 msg = caps.c
705 msg.to = p.from
706 msg.from = p.to.to_s + '/sgx'
707
708 write_to_stream msg
709
710 true
711 end
712
713 disco_items(
714 to: Blather::JID.new(ARGV[0]),
715 node: "http://jabber.org/protocol/commands"
716 ) do |i|
717 fetch_catapult_cred_for(i.from).then { |creds|
718 BandwidthTNOptions.tn_eligible_for_port_out_pin?(creds).then { |eligible|
719 reply = i.reply
720 reply.node = 'http://jabber.org/protocol/commands'
721
722 if eligible
723 reply.items = [
724 Blather::Stanza::DiscoItems::Item.new(
725 i.to,
726 'set-port-out-pin',
727 'Set Port-Out PIN'
728 )
729 ]
730 else
731 reply.items = []
732 end
733
734 write_to_stream reply
735 }
736 }.catch { |e|
737 if e.is_a?(Array) && [2, 3].include?(e.length)
738 write_to_stream i.as_error(e[1], e[0], e[2])
739 else
740 EMPromise.reject(e)
741 end
742 }
743 end
744
745 iq '/iq/ns:query', ns: 'http://jabber.org/protocol/disco#info' do |i|
746 # TODO: return error if i.type is :set - if it is :reply or
747 # :error it should be ignored (as the below does currently);
748 # review specification to see how to handle other type values
749 if i.type != :get
750 LOG.warn("Ignoring non-get disco IQ", type: i.type.to_s, stanza: i.inspect)
751 next
752 end
753
754 # respond to capabilities request for an sgx-bwmsgsv2 number JID
755 if i.to.node
756 # TODO: confirm the node URL is expected using below
757 #puts "XR[node]: #{xpath_result[0]['node']}"
758
759 msg = i.reply
760 msg.node = i.node
761 msg.identities = user_cap_identities
762 msg.features = user_cap_features
763
764 write_to_stream msg
765 next
766 end
767
768 # respond to capabilities request for sgx-bwmsgsv2 itself
769 msg = i.reply
770 msg.node = i.node
771 msg.identities = [{
772 name: 'Soprani.ca Gateway to XMPP - Bandwidth API V2',
773 type: 'sms', category: 'gateway'
774 }]
775 msg.features = @gateway_features
776 write_to_stream msg
777
778 true
779 end
780
781 def self.check_then_register(i, *creds)
782 @registration_repo
783 .put(i.from, *creds)
784 .catch_only(RegistrationRepo::Conflict) { |e|
785 EMPromise.reject([:cancel, 'conflict', e.message])
786 }.then {
787 write_to_stream i.reply
788 }
789 end
790
791 def self.creds_from_registration_query(i)
792 if i.query.find_first("./ns:x", ns: "jabber:x:data")
793 [
794 i.form.field("nick")&.value,
795 i.form.field("username")&.value,
796 i.form.field("password")&.value,
797 i.form.field("phone")&.value
798 ]
799 else
800 [i.nick, i.username, i.password, i.phone]
801 end
802 end
803
804 def self.process_registration(i)
805 EMPromise.resolve(nil).then {
806 if i.remove?
807 @registration_repo.delete(i.from).then do
808 write_to_stream i.reply
809 EMPromise.reject(:done)
810 end
811 else
812 creds_from_registration_query(i)
813 end
814 }.then { |user_id, api_token, api_secret, phone_num|
815 if phone_num && phone_num[0] == '+'
816 [user_id, api_token, api_secret, phone_num]
817 else
818 # TODO: add text re number not (yet) supported
819 EMPromise.reject([:cancel, 'item-not-found'])
820 end
821 }.then { |user_id, api_token, api_secret, phone_num|
822 # TODO: find way to verify #{phone_num}, too
823 call_catapult(
824 api_token,
825 api_secret,
826 :get,
827 "api/v2/users/#{user_id}/media"
828 ).then { |response|
829 JSON.parse(response)
830 # TODO: confirm response is array - could be empty
831
832 LOG.debug("Registration verify response", response: response.to_s[0..999])
833
834 check_then_register(
835 i,
836 user_id,
837 api_token,
838 api_secret,
839 phone_num
840 )
841 }
842 }.catch_only(BandwidthError) { |e|
843 EMPromise.reject(case e.code
844 when 401
845 # TODO: add text re bad credentials
846 [:auth, 'not-authorized']
847 when 404
848 # TODO: add text re number not found or disabled
849 [:cancel, 'item-not-found']
850 else
851 [:modify, 'not-acceptable']
852 end)
853 }
854 end
855
856 def self.registration_form(orig, existing_number=nil)
857 orig.registered = !!existing_number
858
859 # TODO: update "User Id" x2 below (to "accountId"?), and others?
860 orig.instructions = "Enter the information from your Account "\
861 "page as well as the Phone Number\nin your "\
862 "account you want to use (ie. '+12345678901')"\
863 ".\nUser Id is nick, API Token is username, "\
864 "API Secret is password, Phone Number is phone"\
865 ".\n\nThe source code for this gateway is at "\
866 "https://gitlab.com/soprani.ca/sgx-bwmsgsv2 ."\
867 "\nCopyright (C) 2017-2020 Denver Gingerich "\
868 "and others, licensed under AGPLv3+."
869 orig.nick = ""
870 orig.username = ""
871 orig.password = ""
872 orig.phone = existing_number.to_s
873
874 orig.form.fields = [
875 {
876 required: true, type: :"text-single",
877 label: 'User Id', var: 'nick'
878 },
879 {
880 required: true, type: :"text-single",
881 label: 'API Token', var: 'username'
882 },
883 {
884 required: true, type: :"text-private",
885 label: 'API Secret', var: 'password'
886 },
887 {
888 required: true, type: :"text-single",
889 label: 'Phone Number', var: 'phone',
890 value: existing_number.to_s
891 }
892 ]
893 orig.form.title = 'Register for '\
894 'Soprani.ca Gateway to XMPP - Bandwidth API V2'
895 orig.form.instructions = "Enter the details from your Account "\
896 "page as well as the Phone Number\nin your "\
897 "account you want to use (ie. '+12345678901')"\
898 ".\n\nThe source code for this gateway is at "\
899 "https://gitlab.com/soprani.ca/sgx-bwmsgsv2 ."\
900 "\nCopyright (C) 2017-2020 Denver Gingerich "\
901 "and others, licensed under AGPLv3+."
902
903 orig
904 end
905
906 ibr do |i|
907 case i.type
908 when :set
909 process_registration(i)
910 when :get
911 bare_jid = i.from.stripped
912 @registration_repo.find(bare_jid).then { |creds|
913 reply = registration_form(i.reply, creds.last)
914 write_to_stream reply
915 }
916 else
917 # Unknown IQ, ignore for now
918 EMPromise.reject(:done)
919 end.catch { |e|
920 if e.is_a?(Array) && (e.length == 2 || e.length == 3)
921 write_to_stream i.as_error(e[1], e[0], e[2])
922 elsif e != :done
923 EMPromise.reject(e)
924 end
925 }.catch(&method(:panic))
926 end
927
928 command :execute?, node: "set-port-out-pin", sessionid: nil do |iq|
929 # Ensure user is registered, but discard their credentials
930 # because we're just showing them a form.
931 fetch_catapult_cred_for(iq.from).then { |_creds|
932 reply = iq.reply
933 reply.node = 'set-port-out-pin'
934 reply.sessionid = SecureRandom.uuid
935 reply.status = :executing
936
937 form = Blather::Stanza::X.find_or_create(reply.command)
938 form.type = "form"
939 form.fields = [
940 {
941 var: 'pin',
942 type: 'text-private',
943 label: 'Port-Out PIN',
944 required: true
945 },
946 {
947 var: 'confirm_pin',
948 type: 'text-private',
949 label: 'Confirm PIN',
950 required: true
951 }
952 ]
953
954 reply.command.add_child(form)
955 reply.allowed_actions = [:complete]
956
957 write_to_stream reply
958 }.catch { |e|
959 if e.is_a?(Array) && [2, 3].include?(e.length)
960 write_to_stream iq.as_error(e[1], e[0], e[2])
961 else
962 EMPromise.reject(e)
963 end
964 }.catch(&method(:panic))
965 end
966
967 command :complete?, node: "set-port-out-pin", sessionid: /./ do |iq|
968 pin = iq.form.field('pin')&.value
969 confirm_pin = iq.form.field('confirm_pin')&.value
970
971 if pin.nil? || confirm_pin.nil?
972 write_to_stream iq.as_error(
973 'bad-request',
974 :modify,
975 'PIN fields are required'
976 )
977 next
978 end
979
980 if pin != confirm_pin
981 write_to_stream iq.as_error(
982 'bad-request',
983 :modify,
984 'PIN confirmation does not match'
985 )
986 next
987 end
988
989 if pin !~ /\A[a-zA-Z0-9]{4,10}\z/
990 write_to_stream iq.as_error(
991 'bad-request',
992 :modify,
993 'PIN must be 4-10 alphanumeric characters'
994 )
995 next
996 end
997
998 fetch_catapult_cred_for(iq.from).then { |creds|
999 BandwidthTNOptions.set_port_out_pin(creds, pin).then {
1000 reply = iq.reply
1001 reply.node = 'set-port-out-pin'
1002 reply.sessionid = iq.sessionid
1003 reply.status = :completed
1004 reply.note_type = :info
1005 reply.note_text = 'Port-out PIN has been set successfully.'
1006
1007 write_to_stream reply
1008 }.catch { |e|
1009 reply = iq.reply
1010 reply.node = 'set-port-out-pin'
1011 reply.sessionid = iq.sessionid
1012 reply.status = :completed
1013 reply.note_type = :error
1014 error_msg = if e.respond_to?(:message) && e.message.include?('not valid')
1015 "Invalid phone number format. "\
1016 "Please check your registered phone number."
1017 elsif e.respond_to?(:message) && e.message.include?('ErrorCode')
1018 "Bandwidth API error: #{e.message}"
1019 else
1020 "Failed to set port-out PIN. Please try again later."
1021 end
1022 reply.note_text = error_msg
1023
1024 write_to_stream reply
1025 }
1026 }.catch { |e|
1027 if e.is_a?(Array) && [2, 3].include?(e.length)
1028 write_to_stream iq.as_error(e[1], e[0], e[2])
1029 else
1030 EMPromise.reject(e)
1031 end
1032 }.catch(&method(:panic))
1033 end
1034
1035 iq type: [:get, :set] do |iq|
1036 write_to_stream(Blather::StanzaError.new(
1037 iq,
1038 'feature-not-implemented',
1039 :cancel
1040 ))
1041
1042 true
1043 end
1044end
1045
1046class ReceiptMessage < Blather::Stanza
1047 def self.new(to=nil)
1048 node = super :message
1049 node.to = to
1050 node
1051 end
1052end
1053
1054class WebhookHandler < Goliath::API
1055 use Sentry::Rack::CaptureExceptions
1056 use Goliath::Rack::Params
1057
1058 def response(env)
1059 @registration_repo = RegistrationRepo.new
1060 # TODO: add timestamp grab here, and MUST include ./tai version
1061
1062 LOG.debug("Webhook request env", env: env.reject { |k| k == 'params' })
1063
1064 if params.empty?
1065 LOG.warn "Empty webhook params"
1066 return [200, {}, "OK"]
1067 end
1068
1069 if env['REQUEST_URI'] != '/'
1070 LOG.warn("Non-/ request", uri: env['REQUEST_URI'], method: env['REQUEST_METHOD'])
1071 return [200, {}, "OK"]
1072 end
1073
1074 if env['REQUEST_METHOD'] != 'POST'
1075 LOG.warn("Non-POST request", uri: env['REQUEST_URI'], method: env['REQUEST_METHOD'])
1076 return [200, {}, "OK"]
1077 end
1078
1079 # TODO: process each message in list, not just first one
1080 jparams = params.dig('_json', 0, 'message')
1081 type = params.dig('_json', 0, 'type')
1082
1083 return [400, {}, "Missing params\n"] unless jparams && type
1084
1085 users_num, others_num =
1086 if type == 'message-failed' # NOTE: This implies direction == 'out'
1087 [jparams['from'], params['_json'][0]['to']]
1088 elsif jparams['direction'] == 'in'
1089 [jparams['owner'], jparams['from']]
1090 elsif jparams['direction'] == 'out'
1091 [jparams['from'], jparams['owner']] # NOTE: for outbound, 'from' == 'owner'
1092 else
1093 LOG.error("Unexpected message direction", direction: jparams['direction'])
1094 [jparams['from'], jparams['owner']]
1095 end
1096
1097 return [400, {}, "Missing params\n"] unless users_num && others_num
1098 return [400, {}, "Missing params\n"] unless jparams['to'].is_a?(Array)
1099 return [400, {}, "Missing params\n"] if jparams['to'].empty?
1100
1101 LOG.info(
1102 "Webhook message",
1103 message_id: jparams['id'],
1104 event_type: type,
1105 time: jparams['time'],
1106 direction: jparams['direction'],
1107 delivery_state: jparams['deliveryState'],
1108 error_code: jparams['errorCode'],
1109 description: jparams['description'],
1110 tag: jparams['tag'],
1111 media: jparams['media']
1112 )
1113
1114 if others_num[0] != '+'
1115 # TODO: check that others_num actually a shortcode first
1116 others_num +=
1117 ';phone-context=ca-us.phone-context.soprani.ca'
1118 end
1119
1120 bare_jid = @registration_repo.find_jid(users_num).sync
1121
1122 if !bare_jid
1123 LOG.warn("JID not found for number, BW API misconfigured?", users_num: users_num)
1124
1125 return [403, {}, "Customer not found\n"]
1126 end
1127
1128 msg = nil
1129 case jparams['direction']
1130 when 'in'
1131 text = ''
1132 case type
1133 when 'message-received'
1134 # TODO: handle group chat, and fix above
1135 text = jparams['text']
1136
1137 if text.to_s.empty? && Array(jparams['media']).empty?
1138 return [400, {}, "Missing params\n"]
1139 end
1140
1141 if jparams['to'].length > 1
1142 msg = Blather::Stanza::Message.new(
1143 Blather::JID.new(bare_jid).domain
1144 )
1145 msg.body = text unless text&.empty?
1146
1147 addrs = Nokogiri::XML::Node.new(
1148 'addresses', msg.document)
1149 addrs['xmlns'] = 'http://jabber.org/' \
1150 'protocol/address'
1151
1152 addr1 = Nokogiri::XML::Node.new(
1153 'address', msg.document)
1154 addr1['type'] = 'to'
1155 addr1['jid'] = bare_jid
1156 addrs.add_child(addr1)
1157
1158 jparams['to'].reject(
1159 # Don't send to the same person twice,
1160 # and don't send to the person who sent it
1161 &[users_num, others_num].method(:include?)
1162 ).each do |receiver|
1163 addrn = Nokogiri::XML::Node.new(
1164 'address', msg.document)
1165 addrn['type'] = 'to'
1166 addrn['uri'] = "sms:#{receiver}"
1167 addrn['delivered'] = 'true'
1168 addrs.add_child(addrn)
1169 end
1170
1171 msg.add_child(addrs)
1172 end
1173
1174 media_urls = Array(jparams['media']).map { |media_url|
1175 unless media_url.end_with?(
1176 '.smil', '.txt', '.xml'
1177 )
1178 SGXbwmsgsv2.send_media(
1179 others_num + '@' +
1180 ARGV[0],
1181 bare_jid, media_url,
1182 nil, nil, msg
1183 )
1184 media_url
1185 end
1186 }.compact
1187
1188 if text&.empty? || (media_urls.any? && jparams['to'].length > 1)
1189 if !env['HTTP_X_JMP_RESEND_OF'].to_s.empty?
1190 MessageEvent::ResendIn.new(
1191 original_stream_id: env['HTTP_X_JMP_RESEND_OF'],
1192 original_bandwidth_id: jparams['id'],
1193 owner: jparams['owner']
1194 ).emit(REDIS)
1195 else
1196 MessageEvent::In.new(
1197 timestamp: jparams['time'],
1198 from: jparams['from'],
1199 to: jparams['to'],
1200 owner: jparams['owner'],
1201 bandwidth_id: jparams['id'],
1202 body: jparams['text'].to_s,
1203 media_urls: media_urls
1204 ).emit(REDIS)
1205 end
1206
1207 return [200, {}, "OK"]
1208 end
1209 else
1210 text = "unknown type (#{type})" \
1211 " with text: #{jparams['text']}"
1212
1213 # TODO: log/notify of this properly
1214 LOG.warn("Unknown inbound message type", text: text)
1215 end
1216
1217 # If text is not empty, but there isn't a msg,
1218 # we need to construct a msg to convey that text
1219 unless msg || text.to_s.empty?
1220 msg = Blather::Stanza::Message.new(
1221 bare_jid,
1222 # Strip control codes.
1223 # This only happened, or at least caused a problem, once,
1224 # but it seems sensible to say that we shouldn't be getting
1225 # control codes from the PSTN.
1226 text.gsub(/[\u0000-\u0008\u000b-\u001f]/, "")
1227 )
1228 msg.document.encoding = "utf-8"
1229 msg.chat_state = nil
1230 end
1231 else # per prior switch, this is: jparams['direction'] == 'out'
1232 tag_parts = jparams['tag'].split(/ /, 2)
1233 id = WEBrick::HTTPUtils.unescape(tag_parts[0])
1234 resourcepart = WEBrick::HTTPUtils.unescape(tag_parts[1])
1235
1236 # TODO: remove this hack
1237 if jparams['to'].length > 1
1238 case type
1239 when 'message-failed'
1240 MessageEvent::Failed.new(
1241 timestamp: jparams['time'],
1242 stanza_id: id,
1243 bandwidth_id: jparams['id'],
1244 error_code: jparams['errorCode'].to_s,
1245 error_description: jparams['description'].to_s
1246 ).emit(REDIS)
1247 when 'message-delivered'
1248 MessageEvent::Delivered.new(
1249 timestamp: jparams['time'],
1250 stanza_id: id,
1251 bandwidth_id: jparams['id']
1252 ).emit(REDIS)
1253 end
1254 LOG.warn("Group message, skipping receipt", users_num: users_num)
1255 return [200, {}, "OK"]
1256 end
1257
1258 case type
1259 when 'message-failed'
1260 # create a bare message like the one user sent
1261 msg = Blather::Stanza::Message.new(
1262 others_num + '@' + ARGV[0])
1263 msg.from = bare_jid + '/' + resourcepart
1264 msg['id'] = id
1265
1266 # TODO: add 'errorCode' and/or 'description' val
1267 # create an error reply to the bare message
1268 msg = msg.as_error(
1269 'recipient-unavailable',
1270 :wait,
1271 params['_json'][0]['description']
1272 )
1273
1274 when 'message-delivered'
1275
1276 msg = ReceiptMessage.new(bare_jid)
1277
1278 # TODO: put in member/instance variable
1279 msg['id'] = SecureRandom.uuid
1280
1281 # TODO: send only when requested per XEP-0184
1282 rcvd = Nokogiri::XML::Node.new(
1283 'received',
1284 msg.document
1285 )
1286 rcvd['xmlns'] = 'urn:xmpp:receipts'
1287 rcvd['id'] = id
1288 msg.add_child(rcvd)
1289
1290 # TODO: make prettier: this should be done above
1291 others_num = params['_json'][0]['to']
1292 else
1293 # TODO: notify somehow of unknown state receivd?
1294 LOG.warn("Unknown outbound message type", id: id, type: type)
1295 return [200, {}, "OK"]
1296 end
1297
1298 # Keeping this due to the `msg.from=` shuffle just below
1299 LOG.debug("Outbound callback response", stanza: msg.inspect)
1300 end
1301
1302 # if message-failed, we already set msg.from
1303 # moreover, we said `msg = msg.as_error`, and StanzaError
1304 msg.from = others_num + '@' + ARGV[0] if msg.respond_to?(:from=)
1305 SGXbwmsgsv2.write(msg)
1306
1307 # Emit event to messages stream
1308 case [jparams['direction'], type]
1309 when ['in', 'message-received']
1310 if !env['HTTP_X_JMP_RESEND_OF'].to_s.empty?
1311 MessageEvent::ResendIn.new(
1312 original_stream_id: env['HTTP_X_JMP_RESEND_OF'],
1313 original_bandwidth_id: jparams['id'],
1314 owner: jparams['owner']
1315 ).emit(REDIS)
1316 else
1317 media_urls = Array(jparams['media']).reject { |u|
1318 u.end_with?('.smil', '.txt', '.xml')
1319 }
1320 MessageEvent::In.new(
1321 timestamp: jparams['time'],
1322 from: jparams['from'],
1323 to: jparams['to'],
1324 owner: jparams['owner'],
1325 bandwidth_id: jparams['id'],
1326 body: jparams['text'].to_s,
1327 media_urls: media_urls
1328 ).emit(REDIS)
1329 end
1330 when ['out', 'message-failed']
1331 tag_parts = jparams['tag'].split(/ /, 2)
1332 stanza_id = WEBrick::HTTPUtils.unescape(tag_parts[0])
1333 MessageEvent::Failed.new(
1334 timestamp: jparams['time'],
1335 stanza_id: stanza_id,
1336 bandwidth_id: jparams['id'],
1337 error_code: jparams['errorCode'].to_s,
1338 error_description: jparams['description'].to_s
1339 ).emit(REDIS)
1340 when ['out', 'message-delivered']
1341 tag_parts = jparams['tag'].split(/ /, 2)
1342 stanza_id = WEBrick::HTTPUtils.unescape(tag_parts[0])
1343 MessageEvent::Delivered.new(
1344 timestamp: jparams['time'],
1345 stanza_id: stanza_id,
1346 bandwidth_id: jparams['id']
1347 ).emit(REDIS)
1348 end
1349
1350 [200, {}, "OK"]
1351 rescue Exception => e
1352 Sentry.capture_exception(e)
1353 [500, {}, "Error"]
1354 end
1355end
1356
1357at_exit do
1358 LOG.info("Soprani.ca/SMS Gateway for XMPP - Bandwidth API V2",
1359 version: `git rev-parse HEAD`.chomp)
1360
1361 if ARGV.size != 7
1362 warn "Usage: sgx-bwmsgsv2.rb <component_jid> "\
1363 "<component_password> <server_hostname> "\
1364 "<server_port> <application_id> "\
1365 "<http_listen_port> <mms_proxy_prefix_url>"
1366 exit 0
1367 end
1368
1369 LOG.info "Starting"
1370
1371 EM.run do
1372 REDIS = EM::Hiredis.connect
1373
1374 SGXbwmsgsv2.run
1375
1376 # required when using Prosody otherwise disconnects on 6-hour inactivity
1377 EM.add_periodic_timer(3600) do
1378 msg = Blather::Stanza::Iq::Ping.new(:get, 'localhost')
1379 msg.from = ARGV[0]
1380 SGXbwmsgsv2.write(msg)
1381 end
1382
1383 server = Goliath::Server.new('0.0.0.0', ARGV[5].to_i)
1384 server.api = WebhookHandler.new
1385 server.app = Goliath::Rack::Builder.build(server.api.class, server.api)
1386 server.logger = LOG
1387 server.start do
1388 ["INT", "TERM"].each do |sig|
1389 trap(sig) do
1390 EM.defer do
1391 LOG.info "Shutting down gateway"
1392 SGXbwmsgsv2.shutdown
1393
1394 LOG.info "Gateway has terminated"
1395 EM.stop
1396 end
1397 end
1398 end
1399 end
1400 end
1401end unless ENV['ENV'] == 'test'