Filtering Job Arguments (#277)

* Filtering Job Arguments

Adds a new `filter_arguments` option that allows you to filter out sensitive data from job arguments.
This option can be configured globally or per application.

Note: Currently, only root-level hash keys are supported.
If needed, support for key paths, procs, and regular expressions could be added in the future.

* SolidQueue and Resque are the main adapters supported

So remove the limitation from the Readme

* Leftover

* Remove the per-app override

* Specify that are strings

* Move to MissionControl NS

* Don't modify raw_args

They are used for job handling logic e.g. retries.

* Move .first up

* It's args_arr

* Extra newline

* No need for the if

* Extra new line

* No need to memoize a fast operation

* Comment

* Test

* Rewording

* Fix cop

* Rename to extract_args_hash

And return an hash or nil

Author: intrip

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_hash = extract_args_hash(resque_job_hash)
+          ActiveJob::JobProxy.new(args_hash).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_hash(filtered_data)
+              args_hash["arguments"] = MissionControl::Jobs.job_arguments_filter.apply_to(args_hash["arguments"])
+            end
+          end
+        end
+
+        def extract_args_hash(raw_data)
+          (raw_data.dig("payload", "args") || raw_data.dig("args"))&.first
+        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")