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