Feature/serverside encryption

pom.xml

@@ -1,18 +1,18 @@
 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-  <modelVersion>4.0.0</modelVersion>
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
 
-  <groupId>com.tayo</groupId>
-  <artifactId>KinesisEncryption</artifactId>
-  <version>0.0.1-SNAPSHOT</version>
-  <packaging>jar</packaging>
+    <groupId>com.tayo</groupId>
+    <artifactId>KinesisEncryption</artifactId>
+    <version>0.0.1-SNAPSHOT</version>
+    <packaging>jar</packaging>
 
-  <name>KinesisEncryption</name>
-  <url>http://maven.apache.org</url>
+    <name>KinesisEncryption</name>
+    <url>http://maven.apache.org</url>
 
-  <properties>
-    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-  </properties>
+    <properties>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+    </properties>
     <build>
         <plugins>
             <plugin>
@@ -45,64 +45,64 @@
             </repositories>
         </profile>
     </profiles>
-  <dependencies>
-    <dependency>
-      <groupId>junit</groupId>
-      <artifactId>junit</artifactId>
-      <version>3.8.1</version>
-      <scope>test</scope>
-    </dependency>
-    <dependency>
-			<groupId>com.amazonaws</groupId>
-			<artifactId>aws-java-sdk-core</artifactId>
-			<version>1.11.46</version>
-		</dependency>
-		<dependency>
-			<groupId>com.amazonaws</groupId>
-			<artifactId>aws-java-sdk</artifactId>
-			<version>1.11.46</version>
-		</dependency>
-      <dependency>
-          <groupId>org.bouncycastle</groupId>
-          <artifactId>bcprov-ext-jdk15on</artifactId>
-          <version>1.54</version>
-      </dependency>
-		<dependency>
-			<groupId>com.amazonaws</groupId>
-			<artifactId>aws-encryption-sdk-java</artifactId>
-			<version>0.0.1</version>
-		</dependency>
-		 <dependency>
+    <dependencies>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>3.8.1</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.amazonaws</groupId>
+            <artifactId>aws-java-sdk-core</artifactId>
+            <version>1.11.714</version>
+        </dependency>
+        <!--<dependency>
+            <groupId>com.amazonaws</groupId>
+            <artifactId>aws-java-sdk</artifactId>
+            <version>1.11.714</version>
+        </dependency>-->
+        <dependency>
+            <groupId>org.bouncycastle</groupId>
+            <artifactId>bcprov-ext-jdk15on</artifactId>
+            <version>1.54</version>
+        </dependency>
+        <dependency>
+            <groupId>com.amazonaws</groupId>
+            <artifactId>aws-encryption-sdk-java</artifactId>
+            <version>0.0.1</version>
+        </dependency>
+        <dependency>
             <groupId>com.amazonaws</groupId>
             <artifactId>amazon-kinesis-client</artifactId>
-            <version>1.7.2</version>
+            <version>1.13.2</version>
         </dependency>
         <dependency>
             <groupId>com.amazonaws</groupId>
             <artifactId>amazon-kinesis-producer</artifactId>
-            <version>0.12.3</version>
+            <version>0.14.0</version>
         </dependency>
         <dependency>
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-simple</artifactId>
             <version>1.7.13</version>
         </dependency>
         <!-- https://mvnrepository.com/artifact/org.apache.commons/commons-io -->
-   <!-- https://mvnrepository.com/artifact/commons-io/commons-io -->
-<dependency>
-    <groupId>commons-io</groupId>
-    <artifactId>commons-io</artifactId>
-    <version>2.5</version>
-</dependency>
-      <dependency>
-          <groupId>org.json</groupId>
-          <artifactId>json</artifactId>
-          <version>20160212</version>
-      </dependency>
-      <dependency>
-      <groupId>com.googlecode.json-simple</groupId>
-      <artifactId>json-simple</artifactId>
-      <version>1.1.1</version>
-      </dependency>
-  </dependencies>
+        <!-- https://mvnrepository.com/artifact/commons-io/commons-io -->
+        <dependency>
+            <groupId>commons-io</groupId>
+            <artifactId>commons-io</artifactId>
+            <version>2.5</version>
+        </dependency>
+        <dependency>
+            <groupId>org.json</groupId>
+            <artifactId>json</artifactId>
+            <version>20160212</version>
+        </dependency>
+        <dependency>
+            <groupId>com.googlecode.json-simple</groupId>
+            <artifactId>json-simple</artifactId>
+            <version>1.1.1</version>
+        </dependency>
+    </dependencies>
 </project>

src/main/java/kinesisencryption/dao/TickerSalesObject.java

@@ -31,4 +31,41 @@ public String toString()
                 ", timeStamp='" + timeStamp + '\'' +
                 '}';
     }
+
+    public TickerSalesObject()
+    {
+
+    }
+
+    public String getTickerSymbol() {
+        return tickerSymbol;
+    }
+
+    public void setTickerSymbol(String tickerSymbol) {
+        this.tickerSymbol = tickerSymbol;
+    }
+
+    public String getSalesPrice() {
+        return salesPrice;
+    }
+
+    public void setSalesPrice(String salesPrice) {
+        this.salesPrice = salesPrice;
+    }
+
+    public String getOrderId() {
+        return orderId;
+    }
+
+    public void setOrderId(String orderId) {
+        this.orderId = orderId;
+    }
+
+    public String getTimeStamp() {
+        return timeStamp;
+    }
+
+    public void setTimeStamp(String timeStamp) {
+        this.timeStamp = timeStamp;
+    }
 }

src/main/java/kinesisencryption/kcl2/DownstreamThread.java

@@ -0,0 +1,70 @@
+package kinesisencryption.kcl2;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.charset.CharacterCodingException;
+import java.nio.charset.Charset;
+import java.util.*;
+
+import com.fasterxml.jackson.core.JsonParseException;
+import com.fasterxml.jackson.databind.JsonMappingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import kinesisencryption.dao.TickerSalesObject;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.amazonaws.services.kinesis.AmazonKinesisClient;
+import com.amazonaws.services.kinesis.model.Record;
+
+/**
+ * Thread used for decrypting and printing consumed records to the logs
+ *
+ */
+public class DownstreamThread implements Runnable {
+    private static final Logger log = LoggerFactory.getLogger(kinesisencryption.kcl.RecordPrinterThread.class);
+    private List<Record> recordList;
+    private AmazonKinesisClient kinesis;
+
+    public AmazonKinesisClient getKinesis() {
+        return kinesis;
+    }
+
+
+    public List<Record> getRecordList() {
+        return recordList;
+    }
+
+    public DownstreamThread(List<Record> recordList) {
+        this.recordList = recordList;
+    }
+
+
+
+    @Override
+    public void run() {
+        Map <String, Object>resultMap = new HashMap();
+        for (Record record : this.getRecordList()) {
+            try {
+                ByteBuffer buffer = record.getData();
+
+                String result = Charset.forName("UTF-8").newDecoder().decode(buffer).toString();
+                log.info("Cipher Blob :" + record.getData().toString() + " : " + "Decrypted Text is :"
+                        + result);
+                TickerSalesObject salesObject = new TickerSalesObject();
+                ObjectMapper mapper = new ObjectMapper();
+                salesObject = mapper.readValue(result, TickerSalesObject.class);
+                resultMap.put(UUID.randomUUID().toString(), salesObject);
+            } catch (CharacterCodingException e) {
+                log.error("Unable to decode result for " + record.getData().toString() + "with equence number "
+                        + record.getSequenceNumber());
+            } catch (JsonParseException e) {
+                e.printStackTrace();
+            } catch (JsonMappingException e) {
+                e.printStackTrace();
+            } catch (IOException e) {
+                e.printStackTrace();
+            }
+        }
+    }
+}
+

src/main/java/kinesisencryption/kcl2/KCLConsumer.java

@@ -0,0 +1,64 @@
+package kinesisencryption.kcl2;
+
+import com.amazonaws.AmazonClientException;
+import com.amazonaws.auth.AWSCredentialsProvider;
+import com.amazonaws.auth.DefaultAWSCredentialsProviderChain;
+import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorFactory;
+import com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStream;
+import com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration;
+import com.amazonaws.services.kinesis.clientlibrary.lib.worker.Worker;
+import kinesisencryption.utils.KinesisEncryptionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.InetAddress;
+import java.util.UUID;
+
+/**
+ * A consumer application that consumes encrypted records from kinesis streams.
+ */
+public class KCLConsumer {
+    private static final InitialPositionInStream INITIAL_POSITION_IN_STREAM = InitialPositionInStream.TRIM_HORIZON;
+    private static AWSCredentialsProvider credentialsProvider;
+    private static final Logger log = LoggerFactory.getLogger(kinesisencryption.kcl.EncryptedConsumerWithKCL.class);
+
+    private static void initialize() {
+        java.security.Security.setProperty("networkaddress.cache.ttl", "60");
+        credentialsProvider = new DefaultAWSCredentialsProviderChain();
+
+        try {
+            credentialsProvider.getCredentials();
+
+        } catch (Exception e) {
+            throw new AmazonClientException("Cannot find credentials");
+        }
+
+    }
+
+    public static void main(String[] args) throws Exception {
+        initialize();
+
+        String workerId = InetAddress.getLocalHost().getCanonicalHostName() + ":" + UUID.randomUUID();
+        String streamName = KinesisEncryptionUtils.getProperties().getProperty("stream_name");
+        String appName = KinesisEncryptionUtils.getProperties().getProperty("kcl_name");
+        String ddbRegion = KinesisEncryptionUtils.getProperties().getProperty("ddb_region_4_kcl");
+        KinesisClientLibConfiguration kinesisClientLibConfiguration = new KinesisClientLibConfiguration(appName,
+                streamName, credentialsProvider, workerId);
+        kinesisClientLibConfiguration.withInitialPositionInStream(INITIAL_POSITION_IN_STREAM).withRegionName(ddbRegion);
+
+        IRecordProcessorFactory recordProcessorFactory = new KCLRecordProcessorFactory();
+        Worker worker = new Worker(recordProcessorFactory, kinesisClientLibConfiguration);
+
+        log.info("Started KCL Worker process for Stream " + streamName + " " + "with workerId " + workerId);
+
+        int exitCode = 0;
+        try {
+            worker.run();
+        } catch (Throwable t) {
+            System.err.println("Caught throwable while processing data.");
+            t.printStackTrace();
+            exitCode = 1;
+        }
+        System.exit(exitCode);
+    }
+}