Redact passwords from tasks fetched from the TaskQueue by AmatyaAvadhanula · Pull Request #16182 · apache/druid

AmatyaAvadhanula · 2024-03-21T11:59:33Z

Description

Active tasks fetched from TaskQueue's in memory state may reveal sensitive information.

This doesn't happen for payloads fetched from the metadata store, as they are read using a json mapper with an added mixin for redacting credentials.

This PR leans on a similar change in the TaskQueue where a payload is serialized and then deserialized.

Release note

Key changed/added classes in this PR

TaskQueue

This PR has:

kfaraz

Thanks for the fix.
~~I wonder if it would be better to just do the redaction while putting an item in the tasks map. That way the serde would have to be done only once.~~
Edit: On second thought, doing the redaction while putting in the map is not a viable option as this same Task instance is sent across the wire to MMs/indexers.

kfaraz · 2024-03-22T11:51:35Z

  private final TaskActionClientFactory taskActionClientFactory;
  private final TaskLockbox taskLockbox;
  private final ServiceEmitter emitter;
+  private final ObjectMapper PASSWORD_REDACTING_MAPPER;


Rename since it is not a constant anymore.

Suggested change

private final ObjectMapper PASSWORD_REDACTING_MAPPER;

private final ObjectMapper passwordRedactingMapper;

kfaraz · 2024-03-22T12:01:20Z

+        throw DruidException.forPersona(DruidException.Persona.USER)
+                            .ofCategory(DruidException.Category.RUNTIME_FAILURE)
+                            .build(e, "Failed to serialize or deserialize task.");


I think this exception should be targeted either at developer or operator, not the user persona.

Suggested change

throw DruidException.forPersona(DruidException.Persona.USER)

.ofCategory(DruidException.Category.RUNTIME_FAILURE)

.build(e, "Failed to serialize or deserialize task.");

throw DruidException.forPersona(DruidException.Persona.OPERATOR)

.ofCategory(DruidException.Category.RUNTIME_FAILURE)

.build(e, "Failed to serialize or deserialize task[%s].", task.getId());

kfaraz · 2024-03-22T12:01:46Z

        }
      }
-      catch (JsonProcessingException e) {
+      catch (DruidException e) {


Why are we swallowing this exception?

I thought it might be helpful to fallback to the metadata store (TaskStorage) in such a case

We have to do one of two things:
Either (1) Throw JsonProcessingException from TaskQueue.getActiveTask(), catch it here, log it and fallback to using TaskStorage (catch block should have a comment to this effect). In this case, TaskQueue.getActiveTask need not log the exception.
Or (2) Throw DruidException from TaskQueue.getActiveTask() and don't catch it anywhere.

I would prefer the latter as exception during serde shouldn't happen, and it is best to notify the operator about it as soon as possible.
Also if there is an issue deserializing the task, the task storage would be likely to encounter the same issue as it is the same mapper and the same Task object.

kfaraz · 2024-03-22T12:19:49Z

+    taskQueue.start();
+    taskQueue.add(taskWithPassword);
+
+    final Optional<Task> taskFromTaskStorage = taskStorage.getTask(taskWithPassword.getId());


Suggested change

final Optional<Task> taskFromTaskStorage = taskStorage.getTask(taskWithPassword.getId());

final Optional<Task> taskInStorage = taskStorage.getTask(taskWithPassword.getId());

kfaraz · 2024-03-22T12:19:59Z

+    Assert.assertFalse(mapper.writeValueAsString(taskFromTaskStorage.get()).contains(password));
+
+
+    final Optional<Task> taskFromTaskQueue = taskQueue.getActiveTask(taskWithPassword.getId());


Suggested change

final Optional<Task> taskFromTaskQueue = taskQueue.getActiveTask(taskWithPassword.getId());

final Optional<Task> taskInQueue = taskQueue.getActiveTask(taskWithPassword.getId());

kfaraz · 2024-03-22T12:21:50Z

+
+    final Optional<Task> taskFromTaskQueue = taskQueue.getActiveTask(taskWithPassword.getId());
+    Assert.assertTrue(taskFromTaskQueue.isPresent());
+    Assert.assertNotNull(taskFromTaskQueue.get());


Assign the result of optional.get() to a new variable and then do the assertions.

kfaraz · 2024-03-22T15:31:40Z

  private final AtomicInteger statusUpdatesInQueue = new AtomicInteger();
  private final AtomicInteger handledStatusUpdates = new AtomicInteger();

+


Nit: extra line?

AmatyaAvadhanula · 2024-03-23T08:52:27Z

Thank you for the reviews @kfaraz @cryptoe

Redact passwords from tasks fetched from the TaskQueue

0cb3df5

github-actions Bot added the Area - Ingestion label Mar 21, 2024

cryptoe reviewed Mar 22, 2024

View reviewed changes

Comment thread indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java Outdated

kfaraz reviewed Mar 22, 2024

View reviewed changes

Inject mapper and add test

9c50006

AmatyaAvadhanula requested a review from kfaraz March 22, 2024 11:41

kfaraz reviewed Mar 22, 2024

View reviewed changes

Comment thread indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskQueueTest.java Outdated

Address feedback

7dec199

kfaraz reviewed Mar 22, 2024

View reviewed changes

Address feedback

b90d617

kfaraz approved these changes Mar 23, 2024

View reviewed changes

AmatyaAvadhanula merged commit cfa2a90 into apache:master Mar 23, 2024

adarshsanjeev added this to the 30.0.0 milestone May 6, 2024

adarshsanjeev mentioned this pull request May 28, 2024

[DRAFT] 30.0.0 release notes #16505

Closed

	private final ObjectMapper PASSWORD_REDACTING_MAPPER;
	private final ObjectMapper passwordRedactingMapper;

	final Optional<Task> taskFromTaskStorage = taskStorage.getTask(taskWithPassword.getId());
	final Optional<Task> taskInStorage = taskStorage.getTask(taskWithPassword.getId());

		Assert.assertFalse(mapper.writeValueAsString(taskFromTaskStorage.get()).contains(password));


		final Optional<Task> taskFromTaskQueue = taskQueue.getActiveTask(taskWithPassword.getId());

	final Optional<Task> taskFromTaskQueue = taskQueue.getActiveTask(taskWithPassword.getId());
	final Optional<Task> taskInQueue = taskQueue.getActiveTask(taskWithPassword.getId());

		private final AtomicInteger statusUpdatesInQueue = new AtomicInteger();
		private final AtomicInteger handledStatusUpdates = new AtomicInteger();

Conversation

AmatyaAvadhanula commented Mar 21, 2024

Description

Release note

Key changed/added classes in this PR

Uh oh!

Uh oh!

kfaraz left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

AmatyaAvadhanula commented Mar 23, 2024

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

kfaraz left a comment •

edited

Loading