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