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