package backtype.storm.security.serialization;

import java.util.Map;
import org.apache.commons.codec.binary.Hex;
import org.apache.log4j.Logger;

import javax.crypto.KeyGenerator;
import javax.crypto.SecretKey;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.esotericsoftware.kryo.Serializer;
import com.esotericsoftware.kryo.serializers.BlowfishSerializer;

import backtype.storm.serialization.types.ListDelegateSerializer;
import backtype.storm.utils.ListDelegate;
import backtype.storm.Config;

/**
 * Apply Blowfish encrption for tuple communication to bolts
 */
public class BlowfishTupleSerializer extends Serializer<ListDelegate> {
    /**
     * The secret key (if any) for data encryption by blowfish payload serialization factory (BlowfishSerializationFactory). 
     * You should use in via "storm -c topology.tuple.serializer.blowfish.key=YOURKEY -c topology.tuple.serializer=backtype.storm.security.serialization.BlowfishTupleSerializer jar ...".
     */
    public static String SECRET_KEY = "topology.tuple.serializer.blowfish.key";
    private static final Logger LOG = Logger.getLogger(BlowfishSerializer.class);
    private BlowfishSerializer _serializer;

    public BlowfishTupleSerializer(Kryo kryo, Map storm_conf) {
	String encryption_key = null;
	try {
	    encryption_key = (String)storm_conf.get(SECRET_KEY);
	    LOG.debug("Blowfish serializer being constructed ...");
	    if (encryption_key == null) {
		LOG.error("Encryption key not specified");
		throw new RuntimeException("Blowfish encryption key not specified");
	    }
	    byte[] bytes =  Hex.decodeHex(encryption_key.toCharArray());
	    _serializer = new BlowfishSerializer(new ListDelegateSerializer(), bytes);
	} catch (org.apache.commons.codec.DecoderException ex) {
	    LOG.error("Invalid encryption key");
	    throw new RuntimeException("Blowfish encryption key invalid");
	}
    }

    @Override
    public void write(Kryo kryo, Output output, ListDelegate object) {
	_serializer.write(kryo, output, object);
    }

    @Override
    public ListDelegate read(Kryo kryo, Input input, Class<ListDelegate> type) {
	return (ListDelegate)_serializer.read(kryo, input, type);
    }

    /**
     * Produce a blowfish key to be used in "Storm jar" command
     */
    public static void main(String[] args) {
	try{
	    KeyGenerator kgen = KeyGenerator.getInstance("Blowfish");
	    SecretKey skey = kgen.generateKey();
	    byte[] raw = skey.getEncoded();
	    String keyString = new String(Hex.encodeHex(raw));
	    System.out.println("storm -c "+SECRET_KEY+"="+keyString+" -c "+Config.TOPOLOGY_TUPLE_SERIALIZER+"="+BlowfishTupleSerializer.class.getName() + " ..." );
	} catch (Exception ex) {
	    LOG.error(ex.getMessage());
	    ex.printStackTrace();
	}
    }    
}

@@ -1323,17 +1323,24 @@
 conf) %7B%0A
-%09
+        
 String e
@@ -1365,16 +1365,30 @@
 ll;%0A
-%09try %7B%0A%09
+        try %7B%0A        
     
@@ -1440,17 +1440,24 @@
 T_KEY);%0A
-%09
+        
     LOG.
@@ -1504,25 +1504,32 @@
 cted ...%22);%0A
-%09
+        
     if (encr
@@ -1550,18 +1550,32 @@
 null) %7B%0A
-%09%09
+                
 LOG.erro
@@ -1613,10 +1613,24 @@
 %22);%0A
-%09%09
+                
 thro
@@ -1698,16 +1698,30 @@
 %22);%0A
-%09    %7D%0A%09
+            %7D%0A        
     
@@ -1777,25 +1777,32 @@
 arArray());%0A
-%09
+        
     _seriali
@@ -1868,17 +1868,24 @@
 bytes);%0A
-%09
+        
 %7D catch 
@@ -1925,33 +1925,40 @@
 Exception ex) %7B%0A
-%09
+        
     LOG.error(%22I
@@ -1979,20 +1979,31 @@
 ion key%22
-);%0A%09
+, ex);%0A        
     thro
@@ -2058,20 +2058,31 @@
 invalid%22
-);%0A%09
+, ex);%0A        
 %7D%0A    %7D%0A
@@ -2167,16 +2167,20 @@
 ject) %7B%0A
+    
 %09_serial
@@ -2213,16 +2213,16 @@
 bject);%0A
-
     %7D%0A%0A 
@@ -2315,16 +2315,20 @@
 type) %7B%0A
+    
 %09return 
@@ -2378,16 +2378,16 @@
  type);%0A
-
     %7D%0A%0A 
@@ -2514,15 +2514,29 @@
 ) %7B%0A
-%09try%7B%0A%09
+        try%7B%0A        
     
@@ -2593,17 +2593,24 @@
 fish%22);%0A
-%09
+        
     Secr
@@ -2642,21 +2642,28 @@
 eKey();%0A
-%09
+      
     
+  
 byte%5B%5D r
@@ -2686,17 +2686,24 @@
 oded();%0A
-%09
+        
     Stri
@@ -2745,25 +2745,32 @@
 eHex(raw));%0A
-%09
+        
     System.o
@@ -2913,17 +2913,24 @@
 ...%22 );%0A
-%09
+        
 %7D catch 
@@ -2946,17 +2946,24 @@
 n ex) %7B%0A
-%09
+        
     LOG.
@@ -2986,17 +2986,24 @@
 age());%0A
-%09
+        
     ex.p
@@ -3020,17 +3020,24 @@
 race();%0A
-%09
+        
 %7D%0A    %7D 

clean up blowfish code - formatting and include exception cause