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 nil,
829 {'Content-Type' => 'application/json'}
830 ).then { |response|
831 JSON.parse(response)
832 # TODO: confirm response is array - could be empty
833
834 LOG.debug("Registration verify response", response: response.to_s[0..999])
835
836 check_then_register(
837 i,
838 user_id,
839 api_token,
840 api_secret,
841 phone_num
842 )
843 }
844 }.catch_only(BandwidthError) { |e|
845 EMPromise.reject(case e.code
846 when 401
847 # TODO: add text re bad credentials
848 [:auth, 'not-authorized']
849 when 404
850 # TODO: add text re number not found or disabled
851 [:cancel, 'item-not-found']
852 else
853 [:modify, 'not-acceptable']
854 end)
855 }
856 end
857
858 def self.registration_form(orig, existing_number=nil)
859 orig.registered = !!existing_number
860
861 # TODO: update "User Id" x2 below (to "accountId"?), and others?
862 orig.instructions = "Enter the information from your Account "\
863 "page as well as the Phone Number\nin your "\
864 "account you want to use (ie. '+12345678901')"\
865 ".\nUser Id is nick, API Token is username, "\
866 "API Secret is password, Phone Number is phone"\
867 ".\n\nThe source code for this gateway is at "\
868 "https://gitlab.com/soprani.ca/sgx-bwmsgsv2 ."\
869 "\nCopyright (C) 2017-2020 Denver Gingerich "\
870 "and others, licensed under AGPLv3+."
871 orig.nick = ""
872 orig.username = ""
873 orig.password = ""
874 orig.phone = existing_number.to_s
875
876 orig.form.fields = [
877 {
878 required: true, type: :"text-single",
879 label: 'User Id', var: 'nick'
880 },
881 {
882 required: true, type: :"text-single",
883 label: 'API Token', var: 'username'
884 },
885 {
886 required: true, type: :"text-private",
887 label: 'API Secret', var: 'password'
888 },
889 {
890 required: true, type: :"text-single",
891 label: 'Phone Number', var: 'phone',
892 value: existing_number.to_s
893 }
894 ]
895 orig.form.title = 'Register for '\
896 'Soprani.ca Gateway to XMPP - Bandwidth API V2'
897 orig.form.instructions = "Enter the details from your Account "\
898 "page as well as the Phone Number\nin your "\
899 "account you want to use (ie. '+12345678901')"\
900 ".\n\nThe source code for this gateway is at "\
901 "https://gitlab.com/soprani.ca/sgx-bwmsgsv2 ."\
902 "\nCopyright (C) 2017-2020 Denver Gingerich "\
903 "and others, licensed under AGPLv3+."
904
905 orig
906 end
907
908 ibr do |i|
909 case i.type
910 when :set
911 process_registration(i)
912 when :get
913 bare_jid = i.from.stripped
914 @registration_repo.find(bare_jid).then { |creds|
915 reply = registration_form(i.reply, creds.last)
916 write_to_stream reply
917 }
918 else
919 # Unknown IQ, ignore for now
920 EMPromise.reject(:done)
921 end.catch { |e|
922 if e.is_a?(Array) && (e.length == 2 || e.length == 3)
923 write_to_stream i.as_error(e[1], e[0], e[2])
924 elsif e != :done
925 EMPromise.reject(e)
926 end
927 }.catch(&method(:panic))
928 end
929
930 command :execute?, node: "set-port-out-pin", sessionid: nil do |iq|
931 # Ensure user is registered, but discard their credentials
932 # because we're just showing them a form.
933 fetch_catapult_cred_for(iq.from).then { |_creds|
934 reply = iq.reply
935 reply.node = 'set-port-out-pin'
936 reply.sessionid = SecureRandom.uuid
937 reply.status = :executing
938
939 form = Blather::Stanza::X.find_or_create(reply.command)
940 form.type = "form"
941 form.fields = [
942 {
943 var: 'pin',
944 type: 'text-private',
945 label: 'Port-Out PIN',
946 required: true
947 },
948 {
949 var: 'confirm_pin',
950 type: 'text-private',
951 label: 'Confirm PIN',
952 required: true
953 }
954 ]
955
956 reply.command.add_child(form)
957 reply.allowed_actions = [:complete]
958
959 write_to_stream reply
960 }.catch { |e|
961 if e.is_a?(Array) && [2, 3].include?(e.length)
962 write_to_stream iq.as_error(e[1], e[0], e[2])
963 else
964 EMPromise.reject(e)
965 end
966 }.catch(&method(:panic))
967 end
968
969 command :complete?, node: "set-port-out-pin", sessionid: /./ do |iq|
970 pin = iq.form.field('pin')&.value
971 confirm_pin = iq.form.field('confirm_pin')&.value
972
973 if pin.nil? || confirm_pin.nil?
974 write_to_stream iq.as_error(
975 'bad-request',
976 :modify,
977 'PIN fields are required'
978 )
979 next
980 end
981
982 if pin != confirm_pin
983 write_to_stream iq.as_error(
984 'bad-request',
985 :modify,
986 'PIN confirmation does not match'
987 )
988 next
989 end
990
991 if pin !~ /\A[a-zA-Z0-9]{4,10}\z/
992 write_to_stream iq.as_error(
993 'bad-request',
994 :modify,
995 'PIN must be 4-10 alphanumeric characters'
996 )
997 next
998 end
999
1000 fetch_catapult_cred_for(iq.from).then { |creds|
1001 BandwidthTNOptions.set_port_out_pin(creds, pin).then {
1002 reply = iq.reply
1003 reply.node = 'set-port-out-pin'
1004 reply.sessionid = iq.sessionid
1005 reply.status = :completed
1006 reply.note_type = :info
1007 reply.note_text = 'Port-out PIN has been set successfully.'
1008
1009 write_to_stream reply
1010 }.catch { |e|
1011 reply = iq.reply
1012 reply.node = 'set-port-out-pin'
1013 reply.sessionid = iq.sessionid
1014 reply.status = :completed
1015 reply.note_type = :error
1016 error_msg = if e.respond_to?(:message) && e.message.include?('not valid')
1017 "Invalid phone number format. "\
1018 "Please check your registered phone number."
1019 elsif e.respond_to?(:message) && e.message.include?('ErrorCode')
1020 "Bandwidth API error: #{e.message}"
1021 else
1022 "Failed to set port-out PIN. Please try again later."
1023 end
1024 reply.note_text = error_msg
1025
1026 write_to_stream reply
1027 }
1028 }.catch { |e|
1029 if e.is_a?(Array) && [2, 3].include?(e.length)
1030 write_to_stream iq.as_error(e[1], e[0], e[2])
1031 else
1032 EMPromise.reject(e)
1033 end
1034 }.catch(&method(:panic))
1035 end
1036
1037 iq type: [:get, :set] do |iq|
1038 write_to_stream(Blather::StanzaError.new(
1039 iq,
1040 'feature-not-implemented',
1041 :cancel
1042 ))
1043
1044 true
1045 end
1046end
1047
1048class ReceiptMessage < Blather::Stanza
1049 def self.new(to=nil)
1050 node = super :message
1051 node.to = to
1052 node
1053 end
1054end
1055
1056class WebhookHandler < Goliath::API
1057 use Sentry::Rack::CaptureExceptions
1058 use Goliath::Rack::Params
1059
1060 def response(env)
1061 @registration_repo = RegistrationRepo.new
1062 # TODO: add timestamp grab here, and MUST include ./tai version
1063
1064 LOG.debug("Webhook request env", env: env.reject { |k| k == 'params' })
1065
1066 if params.empty?
1067 LOG.warn "Empty webhook params"
1068 return [200, {}, "OK"]
1069 end
1070
1071 if env['REQUEST_URI'] != '/'
1072 LOG.warn("Non-/ request", uri: env['REQUEST_URI'], method: env['REQUEST_METHOD'])
1073 return [200, {}, "OK"]
1074 end
1075
1076 if env['REQUEST_METHOD'] != 'POST'
1077 LOG.warn("Non-POST request", uri: env['REQUEST_URI'], method: env['REQUEST_METHOD'])
1078 return [200, {}, "OK"]
1079 end
1080
1081 # TODO: process each message in list, not just first one
1082 jparams = params.dig('_json', 0, 'message')
1083 type = params.dig('_json', 0, 'type')
1084
1085 return [400, {}, "Missing params\n"] unless jparams && type
1086
1087 users_num, others_num =
1088 if type == 'message-failed' # NOTE: This implies direction == 'out'
1089 [jparams['from'], params['_json'][0]['to']]
1090 elsif jparams['direction'] == 'in'
1091 [jparams['owner'], jparams['from']]
1092 elsif jparams['direction'] == 'out'
1093 [jparams['from'], jparams['owner']] # NOTE: for outbound, 'from' == 'owner'
1094 else
1095 LOG.error("Unexpected message direction", direction: jparams['direction'])
1096 [jparams['from'], jparams['owner']]
1097 end
1098
1099 return [400, {}, "Missing params\n"] unless users_num && others_num
1100 return [400, {}, "Missing params\n"] unless jparams['to'].is_a?(Array)
1101 return [400, {}, "Missing params\n"] if jparams['to'].empty?
1102
1103 LOG.info(
1104 "Webhook message",
1105 message_id: jparams['id'],
1106 event_type: type,
1107 time: jparams['time'],
1108 direction: jparams['direction'],
1109 delivery_state: jparams['deliveryState'],
1110 error_code: jparams['errorCode'],
1111 description: jparams['description'],
1112 tag: jparams['tag'],
1113 media: jparams['media']
1114 )
1115
1116 if others_num[0] != '+'
1117 # TODO: check that others_num actually a shortcode first
1118 others_num +=
1119 ';phone-context=ca-us.phone-context.soprani.ca'
1120 end
1121
1122 bare_jid = @registration_repo.find_jid(users_num).sync
1123
1124 if !bare_jid
1125 LOG.warn("JID not found for number, BW API misconfigured?", users_num: users_num)
1126
1127 return [403, {}, "Customer not found\n"]
1128 end
1129
1130 msg = nil
1131 case jparams['direction']
1132 when 'in'
1133 text = ''
1134 case type
1135 when 'message-received'
1136 # TODO: handle group chat, and fix above
1137 text = jparams['text']
1138
1139 if text.to_s.empty? && Array(jparams['media']).empty?
1140 return [400, {}, "Missing params\n"]
1141 end
1142
1143 if jparams['to'].length > 1
1144 msg = Blather::Stanza::Message.new(
1145 Blather::JID.new(bare_jid).domain
1146 )
1147 msg.body = text unless text&.empty?
1148
1149 addrs = Nokogiri::XML::Node.new(
1150 'addresses', msg.document)
1151 addrs['xmlns'] = 'http://jabber.org/' \
1152 'protocol/address'
1153
1154 addr1 = Nokogiri::XML::Node.new(
1155 'address', msg.document)
1156 addr1['type'] = 'to'
1157 addr1['jid'] = bare_jid
1158 addrs.add_child(addr1)
1159
1160 jparams['to'].reject(
1161 # Don't send to the same person twice,
1162 # and don't send to the person who sent it
1163 &[users_num, others_num].method(:include?)
1164 ).each do |receiver|
1165 addrn = Nokogiri::XML::Node.new(
1166 'address', msg.document)
1167 addrn['type'] = 'to'
1168 addrn['uri'] = "sms:#{receiver}"
1169 addrn['delivered'] = 'true'
1170 addrs.add_child(addrn)
1171 end
1172
1173 msg.add_child(addrs)
1174 end
1175
1176 media_urls = Array(jparams['media']).map { |media_url|
1177 unless media_url.end_with?(
1178 '.smil', '.txt', '.xml'
1179 )
1180 SGXbwmsgsv2.send_media(
1181 others_num + '@' +
1182 ARGV[0],
1183 bare_jid, media_url,
1184 nil, nil, msg
1185 )
1186 media_url
1187 end
1188 }.compact
1189
1190 if text&.empty? || (media_urls.any? && jparams['to'].length > 1)
1191 if !env['HTTP_X_JMP_RESEND_OF'].to_s.empty?
1192 MessageEvent::ResendIn.new(
1193 original_stream_id: env['HTTP_X_JMP_RESEND_OF'],
1194 original_bandwidth_id: jparams['id'],
1195 owner: jparams['owner']
1196 ).emit(REDIS)
1197 else
1198 MessageEvent::In.new(
1199 timestamp: jparams['time'],
1200 from: jparams['from'],
1201 to: jparams['to'],
1202 owner: jparams['owner'],
1203 bandwidth_id: jparams['id'],
1204 body: jparams['text'].to_s,
1205 media_urls: media_urls
1206 ).emit(REDIS)
1207 end
1208
1209 return [200, {}, "OK"]
1210 end
1211 else
1212 text = "unknown type (#{type})" \
1213 " with text: #{jparams['text']}"
1214
1215 # TODO: log/notify of this properly
1216 LOG.warn("Unknown inbound message type", text: text)
1217 end
1218
1219 # If text is not empty, but there isn't a msg,
1220 # we need to construct a msg to convey that text
1221 unless msg || text.to_s.empty?
1222 msg = Blather::Stanza::Message.new(
1223 bare_jid,
1224 # Strip control codes.
1225 # This only happened, or at least caused a problem, once,
1226 # but it seems sensible to say that we shouldn't be getting
1227 # control codes from the PSTN.
1228 text.gsub(/[\u0000-\u0008\u000b-\u001f]/, "")
1229 )
1230 msg.document.encoding = "utf-8"
1231 msg.chat_state = nil
1232 end
1233 else # per prior switch, this is: jparams['direction'] == 'out'
1234 tag_parts = jparams['tag'].split(/ /, 2)
1235 id = WEBrick::HTTPUtils.unescape(tag_parts[0])
1236 resourcepart = WEBrick::HTTPUtils.unescape(tag_parts[1])
1237
1238 # TODO: remove this hack
1239 if jparams['to'].length > 1
1240 case type
1241 when 'message-failed'
1242 MessageEvent::Failed.new(
1243 timestamp: jparams['time'],
1244 stanza_id: id,
1245 bandwidth_id: jparams['id'],
1246 error_code: jparams['errorCode'].to_s,
1247 error_description: jparams['description'].to_s
1248 ).emit(REDIS)
1249 when 'message-delivered'
1250 MessageEvent::Delivered.new(
1251 timestamp: jparams['time'],
1252 stanza_id: id,
1253 bandwidth_id: jparams['id']
1254 ).emit(REDIS)
1255 end
1256 LOG.warn("Group message, skipping receipt", users_num: users_num)
1257 return [200, {}, "OK"]
1258 end
1259
1260 case type
1261 when 'message-failed'
1262 # create a bare message like the one user sent
1263 msg = Blather::Stanza::Message.new(
1264 others_num + '@' + ARGV[0])
1265 msg.from = bare_jid + '/' + resourcepart
1266 msg['id'] = id
1267
1268 # TODO: add 'errorCode' and/or 'description' val
1269 # create an error reply to the bare message
1270 msg = msg.as_error(
1271 'recipient-unavailable',
1272 :wait,
1273 params['_json'][0]['description']
1274 )
1275
1276 when 'message-delivered'
1277
1278 msg = ReceiptMessage.new(bare_jid)
1279
1280 # TODO: put in member/instance variable
1281 msg['id'] = SecureRandom.uuid
1282
1283 # TODO: send only when requested per XEP-0184
1284 rcvd = Nokogiri::XML::Node.new(
1285 'received',
1286 msg.document
1287 )
1288 rcvd['xmlns'] = 'urn:xmpp:receipts'
1289 rcvd['id'] = id
1290 msg.add_child(rcvd)
1291
1292 # TODO: make prettier: this should be done above
1293 others_num = params['_json'][0]['to']
1294 else
1295 # TODO: notify somehow of unknown state receivd?
1296 LOG.warn("Unknown outbound message type", id: id, type: type)
1297 return [200, {}, "OK"]
1298 end
1299
1300 # Keeping this due to the `msg.from=` shuffle just below
1301 LOG.debug("Outbound callback response", stanza: msg.inspect)
1302 end
1303
1304 # if message-failed, we already set msg.from
1305 # moreover, we said `msg = msg.as_error`, and StanzaError
1306 msg.from = others_num + '@' + ARGV[0] if msg.respond_to?(:from=)
1307 SGXbwmsgsv2.write(msg)
1308
1309 # Emit event to messages stream
1310 case [jparams['direction'], type]
1311 when ['in', 'message-received']
1312 if !env['HTTP_X_JMP_RESEND_OF'].to_s.empty?
1313 MessageEvent::ResendIn.new(
1314 original_stream_id: env['HTTP_X_JMP_RESEND_OF'],
1315 original_bandwidth_id: jparams['id'],
1316 owner: jparams['owner']
1317 ).emit(REDIS)
1318 else
1319 media_urls = Array(jparams['media']).reject { |u|
1320 u.end_with?('.smil', '.txt', '.xml')
1321 }
1322 MessageEvent::In.new(
1323 timestamp: jparams['time'],
1324 from: jparams['from'],
1325 to: jparams['to'],
1326 owner: jparams['owner'],
1327 bandwidth_id: jparams['id'],
1328 body: jparams['text'].to_s,
1329 media_urls: media_urls
1330 ).emit(REDIS)
1331 end
1332 when ['out', 'message-failed']
1333 tag_parts = jparams['tag'].split(/ /, 2)
1334 stanza_id = WEBrick::HTTPUtils.unescape(tag_parts[0])
1335 MessageEvent::Failed.new(
1336 timestamp: jparams['time'],
1337 stanza_id: stanza_id,
1338 bandwidth_id: jparams['id'],
1339 error_code: jparams['errorCode'].to_s,
1340 error_description: jparams['description'].to_s
1341 ).emit(REDIS)
1342 when ['out', 'message-delivered']
1343 tag_parts = jparams['tag'].split(/ /, 2)
1344 stanza_id = WEBrick::HTTPUtils.unescape(tag_parts[0])
1345 MessageEvent::Delivered.new(
1346 timestamp: jparams['time'],
1347 stanza_id: stanza_id,
1348 bandwidth_id: jparams['id']
1349 ).emit(REDIS)
1350 end
1351
1352 [200, {}, "OK"]
1353 rescue Exception => e
1354 Sentry.capture_exception(e)
1355 [500, {}, "Error"]
1356 end
1357end
1358
1359at_exit do
1360 LOG.info("Soprani.ca/SMS Gateway for XMPP - Bandwidth API V2",
1361 version: `git rev-parse HEAD`.chomp)
1362
1363 if ARGV.size != 7
1364 warn "Usage: sgx-bwmsgsv2.rb <component_jid> "\
1365 "<component_password> <server_hostname> "\
1366 "<server_port> <application_id> "\
1367 "<http_listen_port> <mms_proxy_prefix_url>"
1368 exit 0
1369 end
1370
1371 LOG.info "Starting"
1372
1373 EM.run do
1374 REDIS = EM::Hiredis.connect
1375
1376 SGXbwmsgsv2.run
1377
1378 # required when using Prosody otherwise disconnects on 6-hour inactivity
1379 EM.add_periodic_timer(3600) do
1380 msg = Blather::Stanza::Iq::Ping.new(:get, 'localhost')
1381 msg.from = ARGV[0]
1382 SGXbwmsgsv2.write(msg)
1383 end
1384
1385 server = Goliath::Server.new('0.0.0.0', ARGV[5].to_i)
1386 server.api = WebhookHandler.new
1387 server.app = Goliath::Rack::Builder.build(server.api.class, server.api)
1388 server.logger = LOG
1389 server.start do
1390 ["INT", "TERM"].each do |sig|
1391 trap(sig) do
1392 EM.defer do
1393 LOG.info "Shutting down gateway"
1394 SGXbwmsgsv2.shutdown
1395
1396 LOG.info "Gateway has terminated"
1397 EM.stop
1398 end
1399 end
1400 end
1401 end
1402 end
1403end unless ENV['ENV'] == 'test'