[HWORKS-556] Remove bijection-avro_2.12 - CVE-2017-15288 (#1398)

featurestore-ee/src/main/java/io/hops/hopsworks/featurestore/databricks/client/DatabricksClient.java

@@ -5,7 +5,6 @@
 package io.hops.hopsworks.featurestore.databricks.client;
 
 import com.damnhandy.uri.template.UriTemplate;
-import com.twitter.bijection.codec.Base64;
 import io.hops.hopsworks.common.proxies.client.HttpClient;
 import io.hops.hopsworks.common.proxies.client.HttpRetryableAction;
 import io.hops.hopsworks.common.proxies.client.NotRetryableClientProtocolException;
@@ -17,6 +16,7 @@
 import org.apache.http.client.methods.HttpGet;
 import org.apache.http.client.methods.HttpPost;
 import org.apache.http.entity.StringEntity;
+import org.glassfish.jersey.internal.util.Base64;
 
 import javax.ejb.ConcurrencyManagement;
 import javax.ejb.ConcurrencyManagementType;
@@ -251,7 +251,7 @@ private void uploadStream(HttpHost dbInstanceHost, DbfsClose dbfsClose, InputStr
     int read = 0;
     while ((read = inputStream.read(data)) > -1) {
       // Send any pending block data
-      sendBlock(dbInstanceHost, token, dbfsClose, Base64.encodeBase64String(Arrays.copyOf(data, read)));
+      sendBlock(dbInstanceHost, token, dbfsClose, Base64.encodeAsString(Arrays.copyOf(data, read)));
     }
   }
 

hopsworks-common/pom.xml

@@ -171,10 +171,6 @@
       <groupId>io.hops.hive</groupId>
       <artifactId>hive-standalone-metastore</artifactId>
     </dependency>
-    <dependency>
-      <groupId>com.twitter</groupId>
-      <artifactId>bijection-avro_2.12</artifactId>
-    </dependency>
     <dependency>
       <groupId>org.freemarker</groupId>
       <artifactId>freemarker</artifactId>

hopsworks-common/src/main/java/io/hops/hopsworks/common/serving/inference/logger/KafkaInferenceLogger.java

@@ -16,8 +16,6 @@
 
 package io.hops.hopsworks.common.serving.inference.logger;
 
-import com.twitter.bijection.Injection;
-import com.twitter.bijection.avro.GenericAvroCodecs;
 import io.hops.hopsworks.common.dao.kafka.KafkaConst;
 import io.hops.hopsworks.common.kafka.KafkaBrokers;
 import io.hops.hopsworks.persistence.entity.project.Project;
@@ -28,7 +26,10 @@
 import io.hops.hopsworks.exceptions.CryptoPasswordNotFoundException;
 import org.apache.avro.Schema;
 import org.apache.avro.generic.GenericData;
-import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.generic.GenericDatumWriter;
+import org.apache.avro.io.DatumWriter;
+import org.apache.avro.io.Encoder;
+import org.apache.avro.io.EncoderFactory;
 import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.ProducerConfig;
@@ -41,6 +42,7 @@
 import javax.ejb.Asynchronous;
 import javax.ejb.EJB;
 import javax.ejb.Stateless;
+import java.io.ByteArrayOutputStream;
 import java.io.File;
 import java.io.IOException;
 import java.util.Properties;
@@ -107,23 +109,24 @@ public void logInferenceRequest(Serving serving, String inferenceRequest,
 
     //Get the schema for the topic and the serializer
     Schema avroSchema = new Schema.Parser().parse(serving.getKafkaTopic().getSubjects().getSchema().getSchema());
-    Injection<GenericRecord, byte[]> recordSerializer = GenericAvroCodecs.toBinary(avroSchema);
-
+
     // Create the GenericRecord from the avroSchema
     GenericData.Record inferenceRecord = new GenericData.Record(avroSchema);
 
     // Populate the Inference Record with data
     populateInfererenceRecord(serving, inferenceRequest, responseHttpCode, inferenceResponse, inferenceRecord,
       schemaVersion);
 
-    // Serialize record to byte
-    byte[] inferenceRecordBytes = recordSerializer.apply(inferenceRecord);
+    // Serialize record to byte array and send it to kafka
+    try (ByteArrayOutputStream out = new ByteArrayOutputStream()) {
+      DatumWriter<GenericData.Record> writer = new GenericDatumWriter<>(avroSchema);
+      Encoder encoder = EncoderFactory.get().binaryEncoder(out, null);
+      writer.write(inferenceRecord, encoder);
+      encoder.flush();
 
-    // Push the record to the topic
-    ProducerRecord<String, byte[]> inferenceKakfaRecord = new ProducerRecord<>(
-        serving.getKafkaTopic().getTopicName(), inferenceRecordBytes);
-
-    try {
+      // Push the record to the topic
+      ProducerRecord<String, byte[]> inferenceKakfaRecord = new ProducerRecord<>(
+          serving.getKafkaTopic().getTopicName(), out.toByteArray());
       kafkaProducer.send(inferenceKakfaRecord);
     } catch (Exception e) {
       LOGGER.log(Level.FINE, "Cannot write to topic: " + serving.getKafkaTopic().getTopicName(), e);

pom.xml

@@ -92,7 +92,6 @@
     <awssdk.version>2.17.19</awssdk.version>
     <bcpkix-jdk15on.version>1.69</bcpkix-jdk15on.version>
     <bcprov-jdk15on.version>1.69</bcprov-jdk15on.version>
-    <bijection-avro_2.12.version>0.9.6</bijection-avro_2.12.version>
     <caffeine.version>2.8.0</caffeine.version>
     <calcite-core.version>1.32.0</calcite-core.version>
     <calcite-server.version>1.32.0</calcite-server.version>
@@ -259,11 +258,6 @@
         <artifactId>oauth2-oidc-sdk</artifactId>
         <version>${oauth2-oidc-sdk.version}</version>
       </dependency>
-      <dependency>
-        <groupId>com.twitter</groupId>
-        <artifactId>bijection-avro_2.12</artifactId>
-        <version>${bijection-avro_2.12.version}</version>
-      </dependency>
       <dependency>
         <groupId>commons-fileupload</groupId>
         <artifactId>commons-fileupload</artifactId>