Merge lp:~henrin10/reversible/reversible into lp:reversible

[Henrik Nergaard]

amqps output support

=== modified file 'anonymiser/UserAuthorisation.py'
--- anonymiser/UserAuthorisation.py	2016-01-03 20:28:12 +0000
+++ anonymiser/UserAuthorisation.py	2016-04-13 15:02:48 +0000
@@ -74,15 +74,17 @@
 	self.cache=None
 	self.producers=[]	# Producer thread(s)
 	self.consumers=[]	# Consumer thread(s)
-        self.makeRequest=XACMLCacheAuthRequest()
+    self.makeRequest=XACMLCacheAuthRequest()
 
     def authoriseDataConsumer(self, SERVICE_ENDPOINT, user, resource, action):
+
         """
         Authorise a new consumer to read the XACML data stream.
         The initial authorisation response contains obligations that
         are used to set up the consumer object.
         FIXME: This belongs in the SOAP stubs section...
         """
+
         # Preferred method is passing in a parsed DOM tree.
         # I use the convenience method that accepts an input string for now.
         # FIXME later: use a validating DOM parser instead of the intermediate solution below.
@@ -91,7 +93,7 @@
         request = self.makeRequest.initialRequest(user, resource, action)
     
         # Evaluate the XACML request 
-	#self.logger.info(request)
+        #self.logger.info(request)
 
         requestHandle=RequestCtx.getInstance(ByteArrayInputStream(request))
         response=self.pdp.evaluate(requestHandle)
@@ -105,59 +107,71 @@
         # Full logic to be added later.
     
         if results[0].DECISIONS[results[0].decision] == "Permit":
-            # OK - I am authorised to the PEP. Set up the caching specification for this data consumer.
+        # OK - I am authorised to the PEP. Set up the caching specification for this data consumer.
     
 	    # Do key handling before any threading starts.
 	    pubkey, privkey = generateRSAKeyPair("Anonymiser")
     	
     
 	    # Set up the decision cache        
-            self.cache=XACMLDecisionCache(self.pdp, results, user)
+        self.cache=XACMLDecisionCache(self.pdp, results, user)
     
-            # Print the caching specification (debugging)
-            #self.logger.info(str(self.cache.cacheSpecification))
+        # Print the caching specification (debugging)
+        # self.logger.info(str(self.cache.cacheSpecification))
     
 	    # Set up the output thread plugin.
 	    se = urlparse(SERVICE_ENDPOINT)
+
 	    if se.scheme == "http":
-		from anonymiserlib.HTTPSOutputThread import HTTPOutputThread
-		outThread = HTTPOutputThread(SERVICE_ENDPOINT)
+            from anonymiserlib.HTTPSOutputThread import HTTPOutputThread
+            outThread = HTTPOutputThread(SERVICE_ENDPOINT)
+
 	    elif se.scheme == "https":
-		from anonymiserlib.HTTPSOutputThread import SSLOutputThread
+            from anonymiserlib.HTTPSOutputThread import SSLOutputThread
 	    	outThread = SSLOutputThread(SERVICE_ENDPOINT)	
+
 	    elif se.scheme == "amqp":
-		from anonymiserlib.RabbitOutputThread import RabbitOutputThread
-	    	outThread = RabbitOutputThread(SERVICE_ENDPOINT)	
+            from anonymiserlib.RabbitOutputThread import RabbitOutputThread
+	    	outThread = RabbitOutputThread(SERVICE_ENDPOINT)
+
 	    elif se.scheme == "amqps":
-		raise NotImplementedException("amqps:// protocol not yet implemented.")
+            from anonymiserlib.RabbitSSLOutputThread import RabbitSSLOutputThread
+            outThread = RabbitSSLOutputThread(SERVICE_ENDPOINT)
+
 	    elif se.scheme == "text":
-		from anonymiserlib.TextOutputThread import TextOutputThread
-		outThread = TextOutputThread(SERVICE_ENDPOINT)   
- 
+		  from anonymiserlib.TextOutputThread import TextOutputThread
+		  outThread = TextOutputThread(SERVICE_ENDPOINT)   
+        
+        else
+            raise NotImplementedException("Error [ " + se.scheme + " ]. This scheme is not implemented, or formatted wrongly")
+
 	    self.producers.append(outThread)
     
 	    # Start the output thread.
 	    outThread.start()
     
-            # Create a new data consumer, read cache spec and set up crypto keys etc.
-            adc=AnonymisedDataConsumer(self.cache, outThread, self.profileData)
+        # Create a new data consumer, read cache spec and set up crypto keys etc.
+        adc=AnonymisedDataConsumer(self.cache, outThread, self.profileData)
     
 	    # Set up crypto specification.
 	    adc.cryptoSpecification(pubkey, privkey, self.cache.keyMap, self.cache.pubKeyMap)
     
-            # Start the data consumer thread
+        # Start the data consumer thread
 	    # ====================================================#
 	    # WARNING: CONSUMER THREAD (ANONOMISER) STARTS HERE.  #
 	    # NO THREAD UNSAFE CODE                               #
 	    # BELOW THIS LINE... OR CODE WILL BREAK...            #
 	    # ====================================================#
-            # Add the data consumer thread to the list of active consumers
-            self.consumers.append(adc)
+
+        # Add the data consumer thread to the list of active consumers
+        self.consumers.append(adc)
+
 	    # and start the thread.
-            adc.start()
+        adc.start()
+
         else:
             # Did not get access.
-            raise AccessDeniedException("Initial authorisation: Access denied for "+user)
+            raise AccessDeniedException("Initial authorisation: Access denied for " + user)
 
         # Return the XACML response for now.
         return results[0].DECISIONS[results[0].decision]

=== modified file 'anonymiser/reversible-anonymiser.sh'
--- anonymiser/reversible-anonymiser.sh	2016-01-03 13:35:06 +0000
+++ anonymiser/reversible-anonymiser.sh	2016-04-13 15:02:48 +0000
@@ -57,6 +57,12 @@
 # PRECYSE ESB server interface
 export CLASSPATH=$CLASSPATH:../idmefStreamServer/lib/precysePublishStreamInterface.jar
 
+# JAR files for AMQPS support
+for MQJAR in ${REVERSIBLE_HOME}/../amqps/*.jar
+do 
+	export CLASSPATH="$CLASSPATH:$MQJAR"
+done 
+
 echo $CLASSPATH
 
 echo "Starting anonymiser, please wait..."