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