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