1# frozen_string_literal: true
2
3require "pg/em/connection_pool"
4require "bigdecimal"
5require "blather/client/dsl" # Require this first to not auto-include
6require "blather/client"
7require "braintree"
8require "date"
9require "dhall"
10require "em-hiredis"
11require "em_promise"
12require "ruby-bandwidth-iris"
13require "sentry-ruby"
14require "statsd-instrument"
15
16Sentry.init
17
18CONFIG =
19 Dhall::Coder
20 .new(safe: Dhall::Coder::JSON_LIKE + [Symbol, Proc])
21 .load(ARGV[0], transform_keys: ->(k) { k&.to_sym })
22
23singleton_class.class_eval do
24 include Blather::DSL
25 Blather::DSL.append_features(self)
26end
27
28require_relative "lib/alt_top_up_form"
29require_relative "lib/add_bitcoin_address"
30require_relative "lib/backend_sgx"
31require_relative "lib/bandwidth_tn_order"
32require_relative "lib/btc_sell_prices"
33require_relative "lib/buy_account_credit_form"
34require_relative "lib/command_list"
35require_relative "lib/customer"
36require_relative "lib/electrum"
37require_relative "lib/em"
38require_relative "lib/payment_methods"
39require_relative "lib/registration"
40require_relative "lib/transaction"
41require_relative "lib/web_register_manager"
42require_relative "lib/statsd"
43
44ELECTRUM = Electrum.new(**CONFIG[:electrum])
45
46Faraday.default_adapter = :em_synchrony
47BandwidthIris::Client.global_options = {
48 account_id: CONFIG[:creds][:account],
49 username: CONFIG[:creds][:username],
50 password: CONFIG[:creds][:password]
51}
52
53def new_sentry_hub(stanza, name: nil)
54 hub = Sentry.get_current_hub&.new_from_top
55 raise "Sentry.init has not been called" unless hub
56
57 hub.push_scope
58 hub.current_scope.clear_breadcrumbs
59 hub.current_scope.set_transaction_name(name) if name
60 hub.current_scope.set_user(jid: stanza.from.stripped.to_s)
61 hub
62end
63
64# Braintree is not async, so wrap in EM.defer for now
65class AsyncBraintree
66 def initialize(environment:, merchant_id:, public_key:, private_key:, **)
67 @gateway = Braintree::Gateway.new(
68 environment: environment,
69 merchant_id: merchant_id,
70 public_key: public_key,
71 private_key: private_key
72 )
73 end
74
75 def respond_to_missing?(m, *)
76 @gateway.respond_to?(m)
77 end
78
79 def method_missing(m, *args)
80 return super unless respond_to_missing?(m, *args)
81
82 EM.promise_defer(klass: PromiseChain) do
83 @gateway.public_send(m, *args)
84 end
85 end
86
87 class PromiseChain < EMPromise
88 def respond_to_missing?(*)
89 false # We don't actually know what we respond to...
90 end
91
92 def method_missing(m, *args)
93 return super if respond_to_missing?(m, *args)
94 self.then { |o| o.public_send(m, *args) }
95 end
96 end
97end
98
99BRAINTREE = AsyncBraintree.new(**CONFIG[:braintree])
100
101def panic(e, hub=nil)
102 m = e.respond_to?(:message) ? e.message : e
103 warn "Error raised during event loop: #{e.class}: #{m}"
104 warn e.backtrace if e.respond_to?(:backtrace)
105 if e.is_a?(::Exception)
106 (hub || Sentry).capture_exception(e, hint: { background: false })
107 else
108 (hub || Sentry).capture_message(e.to_s, hint: { background: false })
109 end
110 exit 1
111end
112
113EM.error_handler(&method(:panic))
114
115when_ready do
116 BLATHER = self
117 REDIS = EM::Hiredis.connect
118 BTC_SELL_PRICES = BTCSellPrices.new(REDIS, CONFIG[:oxr_app_id])
119 DB = PG::EM::ConnectionPool.new(dbname: "jmp")
120 DB.type_map_for_results = PG::BasicTypeMapForResults.new(DB)
121 DB.type_map_for_queries = PG::BasicTypeMapForQueries.new(DB)
122
123 EM.add_periodic_timer(3600) do
124 ping = Blather::Stanza::Iq::Ping.new(:get, CONFIG[:server][:host])
125 ping.from = CONFIG[:component][:jid]
126 self << ping
127 end
128end
129
130# workqueue_count MUST be 0 or else Blather uses threads!
131setup(
132 CONFIG[:component][:jid],
133 CONFIG[:component][:secret],
134 CONFIG[:server][:host],
135 CONFIG[:server][:port],
136 nil,
137 nil,
138 workqueue_count: 0
139)
140
141message to: /\Aaccount@/, body: /./ do |m|
142 StatsD.increment("deprecated_account_bot")
143
144 self << m.reply.tap do |out|
145 out.body = "This bot is deprecated. Please talk to xmpp:cheogram.com"
146 end
147end
148
149before(
150 :iq,
151 type: [:error, :result],
152 to: /\Acustomer_/,
153 from: /(\A|@)#{CONFIG[:sgx]}(\/|\Z)/
154) { |iq| halt if IQ_MANAGER.fulfill(iq) }
155
156before nil, to: /\Acustomer_/, from: /(\A|@)#{CONFIG[:sgx]}(\/|\Z)/ do |s|
157 StatsD.increment("stanza_customer")
158
159 sentry_hub = new_sentry_hub(s, name: "stanza_customer")
160 Customer.for_customer_id(
161 s.to.node.delete_prefix("customer_")
162 ).then { |customer|
163 sentry_hub.current_scope.set_user(
164 id: customer.customer_id,
165 jid: s.from.stripped.to_s
166 )
167 customer.stanza_to(s)
168 }.catch { |e| panic(e, sentry_hub) }
169 halt
170end
171
172ADDRESSES_NS = "http://jabber.org/protocol/address"
173message(
174 to: /\A#{CONFIG[:component][:jid]}\Z/,
175 from: /(\A|@)#{CONFIG[:sgx]}(\/|\Z)/
176) do |m|
177 StatsD.increment("inbound_group_text")
178 sentry_hub = new_sentry_hub(m, name: "message")
179
180 address = m.find("ns:addresses", ns: ADDRESSES_NS).first
181 &.find("ns:address", ns: ADDRESSES_NS)
182 &.find { |el| el["jid"].to_s.start_with?("customer_") }
183 pass unless address
184
185 Customer.for_customer_id(
186 Blather::JID.new(address["jid"].to_s).node.delete_prefix("customer_")
187 ).then(&:jid).then { |customer_jid|
188 m.from = m.from.with(domain: CONFIG[:component][:jid])
189 m.to = m.to.with(domain: customer_jid.domain)
190 address["jid"] = customer_jid.to_s
191 BLATHER << m
192 }.catch { |e| panic(e, sentry_hub) }
193end
194
195# Ignore groupchat messages
196# Especially if we have the component join MUC for notifications
197message(type: :groupchat) { true }
198
199def billable_message(m)
200 (m.body && !m.body.empty?) || m.find("ns:x", ns: OOB.registered_ns).first
201end
202
203message do |m|
204 StatsD.increment("message")
205
206 sentry_hub = new_sentry_hub(m, name: "message")
207 today = Time.now.utc.to_date
208 Customer.for_jid(m.from.stripped).then { |customer|
209 sentry_hub.current_scope.set_user(
210 id: customer.customer_id, jid: m.from.stripped.to_s
211 )
212 EMPromise.all([
213 customer, (customer.incr_message_usage if billable_message(m)),
214 REDIS.exists("jmp_usage_notify-#{customer.customer_id}"),
215 customer.stanza_from(m)
216 ])
217 }.then { |(customer, _, already, _)|
218 next if already == 1
219
220 customer.message_usage((today..(today - 30))).then do |usage|
221 next unless usage > 500
222
223 BLATHER.join(CONFIG[:notify_admin], "sgx-jmp")
224 BLATHER.say(
225 CONFIG[:notify_admin],
226 "#{customer.customer_id} has used #{usage} messages since #{today - 30}",
227 :groupchat
228 )
229 REDIS.set("jmp_usage_notify-#{customer.customer_id}", ex: 60 * 60 * 24)
230 end
231 }.catch { |e| panic(e, sentry_hub) }
232end
233
234message :error? do |m|
235 StatsD.increment("message_error")
236
237 puts "MESSAGE ERROR: #{m.inspect}"
238end
239
240class SessionManager
241 def initialize(blather, id_msg, timeout: 5)
242 @blather = blather
243 @sessions = {}
244 @id_msg = id_msg
245 @timeout = timeout
246 end
247
248 def promise_for(stanza)
249 id = "#{stanza.to.stripped}/#{stanza.public_send(@id_msg)}"
250 @sessions.fetch(id) do
251 @sessions[id] = EMPromise.new
252 EM.add_timer(@timeout) do
253 @sessions.delete(id)&.reject(:timeout)
254 end
255 @sessions[id]
256 end
257 end
258
259 def write(stanza)
260 promise = promise_for(stanza)
261 @blather << stanza
262 promise
263 end
264
265 def fulfill(stanza)
266 id = "#{stanza.from.stripped}/#{stanza.public_send(@id_msg)}"
267 if stanza.error?
268 @sessions.delete(id)&.reject(stanza)
269 else
270 @sessions.delete(id)&.fulfill(stanza)
271 end
272 end
273end
274
275IQ_MANAGER = SessionManager.new(self, :id)
276COMMAND_MANAGER = SessionManager.new(self, :sessionid, timeout: 60 * 60)
277web_register_manager = WebRegisterManager.new
278
279disco_info to: Blather::JID.new(CONFIG[:component][:jid]) do |iq|
280 reply = iq.reply
281 reply.identities = [{
282 name: "JMP.chat",
283 type: "sms",
284 category: "gateway"
285 }]
286 reply.features = [
287 "http://jabber.org/protocol/disco#info",
288 "http://jabber.org/protocol/commands"
289 ]
290 form = Blather::Stanza::X.find_or_create(reply.query)
291 form.type = "result"
292 form.fields = [
293 {
294 var: "FORM_TYPE",
295 type: "hidden",
296 value: "http://jabber.org/network/serverinfo"
297 }
298 ] + CONFIG[:xep0157]
299 self << reply
300end
301
302disco_info do |iq|
303 reply = iq.reply
304 reply.identities = [{
305 name: "JMP.chat",
306 type: "sms",
307 category: "client"
308 }]
309 reply.features = [
310 "urn:xmpp:receipts"
311 ]
312 self << reply
313end
314
315disco_items node: "http://jabber.org/protocol/commands" do |iq|
316 StatsD.increment("command_list")
317
318 sentry_hub = new_sentry_hub(iq, name: iq.node)
319 reply = iq.reply
320
321 CommandList.for(iq.from.stripped).then { |list|
322 reply.items = list.map do |item|
323 Blather::Stanza::DiscoItems::Item.new(
324 iq.to,
325 item[:node],
326 item[:name]
327 )
328 end
329 self << reply
330 }.catch { |e| panic(e, sentry_hub) }
331end
332
333iq "/iq/ns:services", ns: "urn:xmpp:extdisco:2" do |iq|
334 StatsD.increment("extdisco")
335
336 reply = iq.reply
337 reply << Nokogiri::XML::Builder.new {
338 services(xmlns: "urn:xmpp:extdisco:2") do
339 service(
340 type: "sip",
341 host: CONFIG[:sip_host]
342 )
343 end
344 }.doc.root
345
346 self << reply
347end
348
349command :execute?, node: "jabber:iq:register", sessionid: nil do |iq|
350 StatsD.increment("command", tags: ["node:#{iq.node}"])
351
352 sentry_hub = new_sentry_hub(iq, name: iq.node)
353 EMPromise.resolve(nil).then {
354 Customer.for_jid(iq.from.stripped)
355 }.catch {
356 sentry_hub.add_breadcrumb(Sentry::Breadcrumb.new(
357 message: "Customer.create"
358 ))
359 Customer.create(iq.from.stripped)
360 }.then { |customer|
361 sentry_hub.current_scope.set_user(
362 id: customer.customer_id,
363 jid: iq.from.stripped.to_s
364 )
365 sentry_hub.add_breadcrumb(Sentry::Breadcrumb.new(
366 message: "Registration.for"
367 ))
368 Registration.for(
369 iq,
370 customer,
371 web_register_manager
372 ).then(&:write).then { StatsD.increment("registration.completed") }
373 }.catch_only(Blather::Stanza) { |reply|
374 self << reply
375 }.catch { |e| panic(e, sentry_hub) }
376end
377
378def reply_with_note(iq, text, type: :info)
379 reply = iq.reply
380 reply.status = :completed
381 reply.note_type = type
382 reply.note_text = text
383
384 self << reply
385end
386
387# Commands that just pass through to the SGX
388command node: [
389 "number-display",
390 "configure-calls",
391 "record-voicemail-greeting"
392] do |iq|
393 StatsD.increment("command", tags: ["node:#{iq.node}"])
394
395 sentry_hub = new_sentry_hub(iq, name: iq.node)
396 Customer.for_jid(iq.from.stripped).then { |customer|
397 sentry_hub.current_scope.set_user(
398 id: customer.customer_id,
399 jid: iq.from.stripped.to_s
400 )
401
402 customer.stanza_from(iq)
403 }.catch { |e| panic(e, sentry_hub) }
404end
405
406command :execute?, node: "credit cards", sessionid: nil do |iq|
407 StatsD.increment("command", tags: ["node:#{iq.node}"])
408
409 sentry_hub = new_sentry_hub(iq, name: iq.node)
410 reply = iq.reply
411 reply.status = :completed
412
413 Customer.for_jid(iq.from.stripped).then { |customer|
414 oob = OOB.find_or_create(reply.command)
415 oob.url = CONFIG[:credit_card_url].call(
416 reply.to.stripped.to_s.gsub("\\", "%5C"),
417 customer.customer_id
418 )
419 oob.desc = "Manage credits cards and settings"
420
421 reply.note_type = :info
422 reply.note_text = "#{oob.desc}: #{oob.url}"
423
424 self << reply
425 }.catch { |e| panic(e, sentry_hub) }
426end
427
428command :execute?, node: "top up", sessionid: nil do |iq|
429 StatsD.increment("command", tags: ["node:#{iq.node}"])
430
431 sentry_hub = new_sentry_hub(iq, name: iq.node)
432 reply = iq.reply
433 reply.allowed_actions = [:complete]
434
435 Customer.for_jid(iq.from.stripped).then { |customer|
436 BuyAccountCreditForm.for(customer).then do |credit_form|
437 credit_form.add_to_form(reply.form)
438 COMMAND_MANAGER.write(reply).then { |iq2| [customer, credit_form, iq2] }
439 end
440 }.then { |(customer, credit_form, iq2)|
441 iq = iq2 # This allows the catch to use it also
442 Transaction.sale(customer, **credit_form.parse(iq2.form))
443 }.then { |transaction|
444 transaction.insert.then do
445 reply_with_note(iq, "#{transaction} added to your account balance.")
446 end
447 }.catch_only(BuyAccountCreditForm::AmountValidationError) { |e|
448 reply_with_note(iq, e.message, type: :error)
449 }.catch { |e|
450 sentry_hub.capture_exception(e)
451 text = "Failed to buy credit, system said: #{e.message}"
452 reply_with_note(iq, text, type: :error)
453 }.catch { |e| panic(e, sentry_hub) }
454end
455
456command :execute?, node: "alt top up", sessionid: nil do |iq|
457 StatsD.increment("command", tags: ["node:#{iq.node}"])
458
459 sentry_hub = new_sentry_hub(iq, name: iq.node)
460 reply = iq.reply
461 reply.status = :executing
462 reply.allowed_actions = [:complete]
463
464 Customer.for_jid(iq.from.stripped).then { |customer|
465 sentry_hub.current_scope.set_user(
466 id: customer.customer_id,
467 jid: iq.from.stripped.to_s
468 )
469
470 EMPromise.all([AltTopUpForm.for(customer), customer])
471 }.then { |(alt_form, customer)|
472 reply.command << alt_form.form
473
474 COMMAND_MANAGER.write(reply).then do |iq2|
475 AddBitcoinAddress.for(iq2, alt_form, customer).write
476 end
477 }.catch { |e| panic(e, sentry_hub) }
478end
479
480command :execute?, node: "reset sip account", sessionid: nil do |iq|
481 StatsD.increment("command", tags: ["node:#{iq.node}"])
482
483 sentry_hub = new_sentry_hub(iq, name: iq.node)
484 Customer.for_jid(iq.from.stripped).then { |customer|
485 sentry_hub.current_scope.set_user(
486 id: customer.customer_id,
487 jid: iq.from.stripped.to_s
488 )
489 customer.reset_sip_account
490 }.then { |sip_account|
491 reply = iq.reply
492 reply.command << sip_account.form
493 BLATHER << reply
494 }.catch { |e| panic(e, sentry_hub) }
495end
496
497command :execute?, node: "usage", sessionid: nil do |iq|
498 StatsD.increment("command", tags: ["node:#{iq.node}"])
499
500 sentry_hub = new_sentry_hub(iq, name: iq.node)
501 report_for = (Date.today..(Date.today << 1))
502
503 Customer.for_jid(iq.from.stripped).then { |customer|
504 sentry_hub.current_scope.set_user(
505 id: customer.customer_id,
506 jid: iq.from.stripped.to_s
507 )
508
509 customer.usage_report(report_for)
510 }.then { |usage_report|
511 reply = iq.reply
512 reply.status = :completed
513 reply.command << usage_report.form
514 BLATHER << reply
515 }.catch { |e| panic(e, sentry_hub) }
516end
517
518command :execute?, node: "web-register", sessionid: nil do |iq|
519 StatsD.increment("command", tags: ["node:#{iq.node}"])
520
521 sentry_hub = new_sentry_hub(iq, name: iq.node)
522
523 begin
524 jid = iq.form.field("jid")&.value.to_s.strip
525 tel = iq.form.field("tel")&.value.to_s.strip
526 sentry_hub.current_scope.set_user(jid: jid, tel: tel)
527 if iq.from.stripped != CONFIG[:web_register][:from]
528 BLATHER << iq.as_error("forbidden", :auth)
529 elsif jid == "" || tel !~ /\A\+\d+\Z/
530 reply_with_note(iq, "Invalid JID or telephone number.", type: :error)
531 else
532 IQ_MANAGER.write(Blather::Stanza::Iq::Command.new.tap { |cmd|
533 cmd.to = CONFIG[:web_register][:to]
534 cmd.node = "push-register"
535 cmd.form.fields = [var: "to", value: jid]
536 cmd.form.type = "submit"
537 }).then { |result|
538 final_jid = result.form.field("from")&.value.to_s.strip
539 web_register_manager[final_jid] = tel
540 BLATHER << iq.reply.tap { |reply| reply.status = :completed }
541 }.catch { |e| panic(e, sentry_hub) }
542 end
543 rescue StandardError => e
544 sentry_hub.capture_exception(e)
545 end
546end
547
548command sessionid: /./ do |iq|
549 COMMAND_MANAGER.fulfill(iq)
550end
551
552iq type: [:result, :error] do |iq|
553 IQ_MANAGER.fulfill(iq)
554end
555
556iq type: [:get, :set] do |iq|
557 StatsD.increment("unknown_iq")
558
559 self << Blather::StanzaError.new(iq, "feature-not-implemented", :cancel)
560end