Filtering Job Arguments

README.md

@@ -114,6 +114,7 @@ Besides `base_controller_class`, you can also set the following for `MissionCont
 - `scheduled_job_delay_threshold`: the time duration before a scheduled job is considered delayed. Defaults to `1.minute` (a job is considered delayed if it hasn't transitioned from the `scheduled` status 1 minute after the scheduled time).
 - `show_console_help`: whether to show the console help. If you don't want the console help message, set this to `false`—defaults to `true`.
 - `backtrace_cleaner`: a backtrace cleaner used for optionally filtering backtraces on the Failed Jobs detail page. Defaults to `Rails::BacktraceCleaner.new`. See the [Advanced configuration](#advanced-configuration) section for how to configure/override this setting on a per application/server basis.
+- `filter_arguments`: an array of strings representing the job argument keys you want to filter out in the UI. This is useful for hiding sensitive user data. Currently, only root-level hash keys are supported.
 
 This library extends Active Job with a querying interface and the following setting:
 - `config.active_job.default_page_size`: the internal batch size that Active Job will use when sending queries to the underlying adapter and the batch size for the bulk operations defined above—defaults to `1000`.

app/helpers/mission_control/jobs/jobs_helper.rb

@@ -67,7 +67,8 @@ def as_renderable_hash(argument)
       elsif argument["_aj_serialized"]
         ActiveJob::Arguments.deserialize([ argument ]).first
       else
-        argument.without("_aj_symbol_keys", "_aj_ruby2_keywords")
+        MissionControl::Jobs.job_arguments_filter.apply_to(argument)
+          .without("_aj_symbol_keys", "_aj_ruby2_keywords")
           .transform_values { |v| as_renderable_argument(v) }
           .map { |k, v| "#{k}: #{v}" }
           .join(", ")

app/views/mission_control/jobs/jobs/_raw_data.html.erb

@@ -1,4 +1,4 @@
 <h2 class="subtitle">Raw data</h2>
 <pre>
-  <%= JSON.pretty_generate(job.raw_data.without("backtrace")) %>
+  <%= JSON.pretty_generate(job.filtered_raw_data.without("backtrace")) %>
 </pre>

lib/active_job/job_proxy.rb

@@ -7,6 +7,8 @@ class ActiveJob::JobProxy < ActiveJob::Base
   class UnsupportedError < StandardError; end
 
   attr_reader :job_class_name
+  # Raw data with the sensitive user data filtered out.
+  attr_accessor :filtered_raw_data
 
   def initialize(job_data)
     super

lib/active_job/queue_adapters/resque_ext.rb

@@ -185,16 +185,29 @@ def fetch_queue_resque_jobs
         end
 
         def deserialize_resque_job(resque_job_hash, index)
-          args_hash = resque_job_hash.dig("payload", "args") || resque_job_hash.dig("args")
-          ActiveJob::JobProxy.new(args_hash&.first).tap do |job|
+          args_arr = extract_args_arr(resque_job_hash)
+          ActiveJob::JobProxy.new(args_arr&.first).tap do |job|
             job.last_execution_error = execution_error_from_resque_job(resque_job_hash)
             job.raw_data = resque_job_hash
+            job.filtered_raw_data = filter_raw_data_arguments(resque_job_hash)
             job.position = jobs_relation.offset_value + index
             job.failed_at = resque_job_hash["failed_at"]&.to_datetime&.utc
             job.status = job.failed_at.present? ? :failed : :pending
           end
         end
 
+        def filter_raw_data_arguments(raw_data)
+          raw_data.deep_dup.tap do |filtered_data|
+            if args_hash = extract_args_arr(filtered_data)&.first
+              args_hash["arguments"] = MissionControl::Jobs.job_arguments_filter.apply_to(args_hash["arguments"])
+            end
+          end
+        end
+
+        def extract_args_arr(raw_data)
+          raw_data.dig("payload", "args") || raw_data.dig("args")
+        end
+
         def execution_error_from_resque_job(resque_job_hash)
           if resque_job_hash["exception"].present?
             ActiveJob::ExecutionError.new \

lib/active_job/queue_adapters/solid_queue_ext.rb

@@ -99,6 +99,7 @@ def deserialize_and_proxy_solid_queue_job(solid_queue_job, job_status = nil)
         job.status = job_status
         job.last_execution_error = execution_error_from_solid_queue_job(solid_queue_job) if job_status == :failed
         job.raw_data = solid_queue_job.as_json
+        job.filtered_raw_data = filter_raw_data_arguments(job.raw_data)
         job.failed_at = solid_queue_job&.failed_execution&.created_at if job_status == :failed
         job.finished_at = solid_queue_job.finished_at
         job.blocked_by = solid_queue_job.concurrency_key
@@ -109,6 +110,12 @@ def deserialize_and_proxy_solid_queue_job(solid_queue_job, job_status = nil)
       end
     end
 
+    def filter_raw_data_arguments(raw_data)
+      raw_data.deep_dup.tap do |filtered_raw_data|
+        filtered_raw_data["arguments"]["arguments"] = MissionControl::Jobs.job_arguments_filter.apply_to(filtered_raw_data.dig("arguments", "arguments"))
+      end
+    end
+
     def status_from_solid_queue_job(solid_queue_job)
       SolidQueueJobs::STATUS_MAP.invert[solid_queue_job.status]
     end

lib/mission_control/jobs.rb

@@ -31,5 +31,11 @@ module Jobs
     mattr_accessor :http_basic_auth_user
     mattr_accessor :http_basic_auth_password
     mattr_accessor :http_basic_auth_enabled, default: true
+
+    mattr_accessor :filter_arguments, default: []
+
+    def self.job_arguments_filter
+      MissionControl::Jobs::ArgumentsFilter.new(filter_arguments)
+    end
   end
 end

lib/mission_control/jobs/arguments_filter.rb

@@ -0,0 +1,24 @@
+# Replaces argument values with [FILTERED] for any keys that match a filter.
+class MissionControl::Jobs::ArgumentsFilter
+  FILTERED = "[FILTERED]"
+
+  def initialize(filter)
+    @filter = filter
+  end
+
+  def apply_to(arguments)
+    case arguments
+    when Array
+      arguments.map { |a| apply_to(a) }
+    when Hash
+      arguments.map do |k, v|
+        [k, filter.include?(k.to_s) ? FILTERED : v]
+      end.to_h
+    else
+      arguments
+    end
+  end
+
+  private
+    attr_reader :filter
+end

test/active_job/queue_adapters/adapter_testing/retry_jobs.rb

@@ -116,4 +116,24 @@ module ActiveJob::QueueAdapters::AdapterTesting::RetryJobs
       failed_job.retry
     end
   end
+
+  test "retrying a single job with filtered arguments preserves the original arguments" do
+    @previous_filter_arguments, MissionControl::Jobs.filter_arguments = MissionControl::Jobs.filter_arguments, %w[ author ]
+    arguments = [ Post.create(title: "hello_world"), 1.year.ago, { author: "Jorge", price: 10 } ]
+    FailingPostJob.perform_later(arguments)
+    perform_enqueued_jobs
+
+    failed_job = ActiveJob.jobs.failed.last
+    failed_job.retry
+
+    perform_enqueued_jobs
+
+    invocations = FailingPostJob.invocations
+    assert_equal 2, invocations.count
+    invocations.each do |invocation|
+      assert_equal arguments, invocation.arguments.first
+    end
+  ensure
+    MissionControl::Jobs.filter_arguments = @previous_filter_arguments
+  end
 end

test/dummy/config/initializers/mission_control_jobs.rb

@@ -15,6 +15,9 @@ def redis_connection_for(app, server)
   Resque::DataStore.new redis_namespace
 end
 
+# Filter sensitive arguments from the UI.
+MissionControl::Jobs.filter_arguments = %w[ author ]
+
 SERVERS_BY_APP.each do |app, servers|
   queue_adapters_by_name = servers.collect do |server|
     queue_adapter = if server.start_with?("resque")

test/dummy/db/seeds.rb

@@ -46,7 +46,7 @@ def load_finished_jobs
 
     def load_failed_jobs
       puts "Generating #{failed_jobs_count} failed jobs for #{application} - #{server}..."
-      failed_jobs_count.times { |index| enqueue_one_of FailingJob => index, FailingReloadedJob => index, FailingPostJob => [ Post.last, 1.year.ago ] }
+      failed_jobs_count.times { |index| enqueue_one_of FailingJob => index, FailingReloadedJob => index, FailingPostJob => [ Post.last, 1.year.ago, author: "Jorge" ] }
       perform_jobs
     end
 

test/mission_control/jobs/arguments_filter_test.rb

@@ -0,0 +1,28 @@
+require "test_helper"
+
+class MissionControl::Jobs::ArgumentsFilterTest < ActiveSupport::TestCase
+  test "apply_to array" do
+    arguments = [
+      "deliver",
+      {
+        email_address: "jorge@37signals.com",
+        profile: { name: "Jorge Manrubia" },
+        message: "Hello!"
+      }
+    ]
+    filtered = MissionControl::Jobs::ArgumentsFilter.new(%w[ email_address message ]).apply_to(arguments)
+
+    assert_equal "deliver", filtered[0]
+    assert_equal({ email_address: "[FILTERED]", profile: { name: "Jorge Manrubia" }, message: "[FILTERED]" }, filtered[1])
+  end
+
+  test "apply_to hash" do
+    argument = {
+      email_address: "jorge@37signals.com",
+      message: "Hello!"
+    }
+    filtered = MissionControl::Jobs::ArgumentsFilter.new(%w[ message ]).apply_to(argument)
+
+    assert_equal({ email_address: "jorge@37signals.com", message: "[FILTERED]" }, filtered)
+  end
+end

test/system/show_failed_job_test.rb

@@ -16,6 +16,22 @@ class ShowFailedJobsTest < ApplicationSystemTestCase
     assert_text /failing_job.rb/
   end
 
+  test "filtered arguments are hidden" do
+    ActiveJob.jobs.failed.discard_all
+    FailingPostJob.perform_later(Post.create(title: "hello_world"), 1.year.ago, author: "Jorge")
+    perform_enqueued_jobs
+    @previous_filter_arguments, MissionControl::Jobs.filter_arguments = MissionControl::Jobs.filter_arguments, %w[ author ]
+
+    visit jobs_path(:failed)
+    click_on "FailingPostJob"
+
+    assert_text /dummy\/post/i
+    assert_text /\[FILTERED\]/
+    assert_no_text /Jorge/
+  ensure
+    MissionControl::Jobs.filter_arguments = @previous_filter_arguments
+  end
+
   test "click on a failed job error to see its error information" do
     within_job_row /FailingJob\s*2/ do
       click_on "RuntimeError: This always fails!"