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