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