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