KAFKA-20450 : Fix : Allowlist based SafeObjectInputStream

[subbudvk]

The current SafeObjectInputStream uses a denylist based approach -
having a fixed denylist to be validated against for deserialization.
This is a bad security practise and has also been advised so in the
original PR.

Making this as a allowlist instead and allowing safe BASE_TYPES which
are required by current caller
(org.apache.kafka.connect.storage.FileOffsetBackingStore)

Also providing a SafeObjectInputStream(InputStream in, Set<String> allowedClasses) so if any consumer require any specific allowedClasses
they can pass in here.

[chia7712]

Please see the discussion on #22048, which we don't think this is a security issue :)

[mimaison]

Even tough we don't consider it a security issue, some cleanup in that area could be useful. I still need to check the PR but I'd be open to improving SafeObjectInputStream so 1) it's overall safer/cleaner code and 2) we stop receiving reports for it!

[subbudvk]

Even tough we don't consider it a security issue, some cleanup in that area could be useful. I still need to check the PR but I'd be open to improving SafeObjectInputStream so 2) it's overall safer/cleaner code and 2) we stop receiving reports for it!

+1

1. Currently SafeObjectInputStream is not really safe since it uses a black list.
2. It currently is used in one consumer only, exploitation on the path requires file access so it's considered fine for now but the utility needs to be modified so that it serves it the purpose it advertises to do. Because the denylist is anyway not useful.

[mimaison]

The PR is failing spotless:

Execution failed for task ':connect:runtime:spotlessJavaCheck'.
> The following files had format violations:
      src/main/java/org/apache/kafka/connect/util/SafeObjectInputStream.java
          @@ -16,9 +16,9 @@
           ·*/
           package·org.apache.kafka.connect.util;
           
          -import·java.io.InvalidClassException;
           import·java.io.IOException;
           import·java.io.InputStream;
          +import·java.io.InvalidClassException;
           import·java.io.ObjectInputStream;
           import·java.io.ObjectStreamClass;
           import·java.util.Objects;
  Run './gradlew spotlessApply' to fix all violations.

[chia7712]

since it uses a black list.

should we worry about the compatibility issue?

[subbudvk]

The PR is failing spotless:

Execution failed for task ':connect:runtime:spotlessJavaCheck'.
> The following files had format violations:
      src/main/java/org/apache/kafka/connect/util/SafeObjectInputStream.java
          @@ -16,9 +16,9 @@
           ·*/
           package·org.apache.kafka.connect.util;
           
          -import·java.io.InvalidClassException;
           import·java.io.IOException;
           import·java.io.InputStream;
          +import·java.io.InvalidClassException;
           import·java.io.ObjectInputStream;
           import·java.io.ObjectStreamClass;
           import·java.util.Objects;
  Run './gradlew spotlessApply' to fix all violations.

@mimaison corrected