Change to StringBuilder, add more on security

Fix tests, clean up

Author: rob42

TODO.txt

@@ -30,7 +30,7 @@ Sailgauge tries port 8080
 Make Waypoint 'here' option
 
 Enable users and passwords and protect server pages
-Add restart from web page
+
 config.html wont delete arrays
 NMEA ais outputs incorrect position format?
 ais requires Cog true? cogM not considered.
@@ -40,10 +40,10 @@ log4j and config may need re-working after upgrade.
 REST api needs attention - http://www:8088/signalk/v1/config returns all
 
 Finish adding XMPP support
-	start xmpp routes
 	add outgoing filters
 	use queue to collect buffer
 	* dont make xmmp.42.co.nz the default!
+	*firewall needs an MSG_SRC_IP or a filter on XMPP server.
 
 need avahi-utils?
 

src/main/java/nz/co/fortytwo/signalk/processor/FullExportProcessor.java

@@ -227,9 +227,9 @@ public void recordEvent(PathEvent pathEvent) {
             return;
 
         if(path.endsWith(dot+source)
-    			&& path.endsWith(dot+timestamp)
-    			&& path.contains(dot+source+dot)
-    			&& path.endsWith(dot+sourceRef)){
+    			|| path.endsWith(dot+timestamp)
+    			|| path.contains(dot+source+dot)
+    			|| path.endsWith(dot+sourceRef)){
         	return;
         }
         // Send update if necessary.

src/main/java/nz/co/fortytwo/signalk/processor/IncomingSecurityFirewall.java

@@ -23,30 +23,30 @@
  */
 package nz.co.fortytwo.signalk.processor;
 
-import static nz.co.fortytwo.signalk.util.SignalKConstants.CONFIG;
 import static nz.co.fortytwo.signalk.util.SignalKConstants.CONFIG_ACTION;
 import static nz.co.fortytwo.signalk.util.SignalKConstants.CONFIG_ACTION_SAVE;
 import static nz.co.fortytwo.signalk.util.SignalKConstants.CONTEXT;
+import static nz.co.fortytwo.signalk.util.SignalKConstants.DEMO;
 import static nz.co.fortytwo.signalk.util.SignalKConstants.INTERNAL_IP;
 import static nz.co.fortytwo.signalk.util.SignalKConstants.MSG_SRC_IP;
 import static nz.co.fortytwo.signalk.util.SignalKConstants.MSG_TYPE;
 import static nz.co.fortytwo.signalk.util.SignalKConstants.PUT;
 import static nz.co.fortytwo.signalk.util.SignalKConstants.SERIAL;
 import static nz.co.fortytwo.signalk.util.SignalKConstants.UPDATES;
-import static nz.co.fortytwo.signalk.util.SignalKConstants.dot;
 import static nz.co.fortytwo.signalk.util.SignalKConstants.self;
 
 import java.util.ArrayList;
 import java.util.List;
 
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
 import mjson.Json;
 import nz.co.fortytwo.signalk.util.ConfigConstants;
 import nz.co.fortytwo.signalk.util.Util;
 
-import org.apache.camel.Exchange;
-import org.apache.camel.Processor;
-import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger;
-
 /**
  * Parse the signalkModel json and remove anything that violates security
  * 
@@ -90,16 +90,21 @@ public void process(Exchange exchange) throws Exception {
 			// we trust local serial
 			String type = exchange.getIn().getHeader(MSG_TYPE,
 					String.class);
-			if (SERIAL.equals(type))
+			if (SERIAL.equals(type)||DEMO.equals(type)){
+				if (logger.isDebugEnabled())
+					logger.debug("Accepting msg:" + type);
 				return;
-
+			}
+			//TODO: check for valid XMPP/STOMP/MQTT SRC_IP ??
+			
 			// we filter on ip
 			String srcIp = exchange.getIn().getHeader(MSG_SRC_IP,
 					String.class);
 			if (logger.isDebugEnabled())
 				logger.debug("Checking src ip:" + srcIp);
 			if (srcIp == null) {
-				logger.debug(exchange);
+				logger.debug("Src ip null:"+exchange.getIn().getHeaders());
+				//exchange.getIn().setBody(null);
 				return;
 			}
 			// denied - drop now
@@ -131,8 +136,7 @@ public void process(Exchange exchange) throws Exception {
 			// we trust INTERNAL_IP
 			if (INTERNAL_IP.equals(type)) {
 				if (logger.isDebugEnabled())
-					logger.debug("Message allowed for src ip (internal):"
-							+ srcIp);
+					logger.debug("Message allowed for src ip (internal):"+ srcIp+":"+exchange.getIn().getHeaders());
 				return;
 			}
 

src/main/java/nz/co/fortytwo/signalk/processor/SourceToSourceRefProcessor.java

@@ -24,24 +24,22 @@
 
 package nz.co.fortytwo.signalk.processor;
 
+import static nz.co.fortytwo.signalk.util.SignalKConstants.MSG_TYPE;
 import static nz.co.fortytwo.signalk.util.SignalKConstants.dot;
-import static nz.co.fortytwo.signalk.util.SignalKConstants.normal;
+import static nz.co.fortytwo.signalk.util.SignalKConstants.label;
 import static nz.co.fortytwo.signalk.util.SignalKConstants.source;
 import static nz.co.fortytwo.signalk.util.SignalKConstants.sourceRef;
 import static nz.co.fortytwo.signalk.util.SignalKConstants.sources;
-import static nz.co.fortytwo.signalk.util.SignalKConstants.type;
-import static nz.co.fortytwo.signalk.util.SignalKConstants.vessels_dot_self_dot;
 
 import java.util.Map.Entry;
 import java.util.NavigableMap;
 
-import mjson.Json;
-import nz.co.fortytwo.signalk.model.SignalKModel;
-import nz.co.fortytwo.signalk.util.JsonSerializer;
-
 import org.apache.camel.Exchange;
 import org.apache.camel.Processor;
-import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import nz.co.fortytwo.signalk.model.SignalKModel;
 
 /**
  * Replaces source with the actual $sourceRef object and stores source in sources.*
@@ -66,27 +64,28 @@ public void process(Exchange exchange) throws Exception {
 			for(String key : model.getKeys()){
 				//get the source.type key
 				//if(logger.isDebugEnabled())logger.debug("Key:"+key);
-				if(key.endsWith(dot+source+dot+type)){
+				if(key.endsWith(dot+source+dot+label)){
 					if(logger.isDebugEnabled())logger.debug("Convert key:"+key);
 					int pos = key.indexOf(dot+source+dot);
-					String typeVal = (String) model.get(key);
+					String typeVal = exchange.getIn().getHeader(MSG_TYPE,String.class);
 
 					int pos1 = pos+source.length()+2;
 					String refKey = key.substring(0,pos);
-					if(logger.isDebugEnabled())logger.debug("refKey:"+refKey+", type:"+typeVal);
+					if(logger.isDebugEnabled())logger.debug("refKey:"+refKey+", bus:"+typeVal);
 					//get the label
-					String label = (String) model.get(refKey+dot+source+dot+"label");
+					String lbl = (String) model.get(refKey+dot+source+dot+label);
+					if(logger.isDebugEnabled())logger.debug("refKey:"+refKey+", label:"+lbl);
 					//set sourceRef
-					model.getFullData().put(refKey+dot+sourceRef, typeVal+dot+label);
+					model.getFullData().put(refKey+dot+sourceRef, typeVal+dot+lbl);
 					//put in sources
 					NavigableMap<String, Object> node = signalkModel.getSubMap(refKey+dot+source);
 					if(logger.isDebugEnabled())logger.debug("Found keys:"+node.size());
 					//node is the source object 
 					if(node!=null){
 						for(Entry<String, Object> entry:node.entrySet()){
 							String nodeKey = entry.getKey().substring(pos1);
-							model.getFullData().put(sources+dot+typeVal+dot+label+dot+nodeKey, entry.getValue());
-							if(logger.isDebugEnabled())logger.debug("Added key:"+sources+dot+typeVal+dot+label+dot+nodeKey+"="+entry.getValue());
+							model.getFullData().put(sources+dot+typeVal+dot+lbl+dot+nodeKey, entry.getValue());
+							if(logger.isDebugEnabled())logger.debug("Added key:"+sources+dot+typeVal+dot+lbl+dot+nodeKey+"="+entry.getValue());
 						}
 
 						//drop the source key and all subkeys

src/main/java/nz/co/fortytwo/signalk/server/RouteManager.java

@@ -24,9 +24,12 @@
 package nz.co.fortytwo.signalk.server;
 
 import static nz.co.fortytwo.signalk.util.ConfigConstants.OUTPUT_XMPP;
+import static nz.co.fortytwo.signalk.util.SignalKConstants.DEMO;
 //import static nz.co.fortytwo.signalk.util.ConfigConstants.UUID;
 import static nz.co.fortytwo.signalk.util.SignalKConstants.FORMAT_DELTA;
 import static nz.co.fortytwo.signalk.util.SignalKConstants.MSG_SRC_BUS;
+import static nz.co.fortytwo.signalk.util.SignalKConstants.MSG_SRC_IP;
+import static nz.co.fortytwo.signalk.util.SignalKConstants.MSG_TYPE;
 import static nz.co.fortytwo.signalk.util.SignalKConstants.POLICY_IDEAL;
 import static nz.co.fortytwo.signalk.util.SignalKConstants.SIGNALK_API;
 import static nz.co.fortytwo.signalk.util.SignalKConstants.SIGNALK_AUTH;
@@ -154,7 +157,7 @@ public void configure0() throws Exception {
 		        .redeliveryDelay(1000));
 
 		from ("direct:fail").id("Fail")
-        .to("log:log:nz.co.fortytwo.signalk.error?level=ERROR&showAll=true");
+        .to("log:nz.co.fortytwo.signalk.error?level=ERROR&showAll=true");
 
 		SignalKModelFactory.load(signalkModel);
 
@@ -421,9 +424,11 @@ public void configure0() throws Exception {
 				.onException(Exception.class).handled(true).maximumRedeliveries(0)
 				.to("log:nz.co.fortytwo.signalk.model.receive?level=ERROR&showException=true&showStackTrace=true")
 				.end()
-			.split(body().tokenize("\n")).streaming()
+			.split(body().regexTokenize("[\r\n|\n|\r]")).streaming()
 			.convertBodyTo(String.class)
-			.throttle(2).timePeriodMillis(1000)
+			.throttle(4).timePeriodMillis(1000)
+			.setHeader(MSG_TYPE, constant(DEMO))
+			//.setHeader(MSG_SRC_IP,constant("127.0.0.1"))
 			.setHeader(MSG_SRC_BUS, constant("demo"))
 			.to(SEDA_INPUT).id(SignalkRouteFactory.getName("SEDA_INPUT"))
 			.end();

src/main/java/nz/co/fortytwo/signalk/server/SerialPortReader.java

@@ -155,7 +155,7 @@ public class SerialReader implements SerialPortEventListener {
 
 		private Pattern uid;
 		List<String> lines = new ArrayList<String>();
-		StringBuffer line = new StringBuffer(60);
+		StringBuilder line = new StringBuilder(60);
 		private boolean enableSerial=true;
 		private boolean complete;
 		protected InputStream in;
@@ -194,7 +194,7 @@ public void serialEvent(SerialPortEvent event) {
 								//10=LF, 13=CR, lines should end in CR/LF
 								if(r==10  ||x==256){
 									if(r==10)complete=true;
-									line.append(new String(buff));
+									line.append(buff);
 									buff=new byte[256];
 									x=0;
 								}
@@ -239,7 +239,7 @@ public void serialEvent(SerialPortEvent event) {
 									}
 								}
 								complete=false;
-								line=new StringBuffer(60);
+								line=new StringBuilder(60);
 							}
 						}
 				}

src/main/java/nz/co/fortytwo/signalk/server/SubscriptionManager.java

@@ -50,7 +50,9 @@ public class SubscriptionManager {
 
 	//hold sessionid <> wsSessionId
 	BiMap<String, String> wsSessionMap = HashBiMap.create();
+	//map wsSession to output type
 	Map<String, String> outPutMap = new HashMap<String, String>();
+	//wsSession> localIp#remoteIp
 	Map<String, String> ipMap = new HashMap<String, String>();
 	//wsSessionId>Subscription
 	ConcurrentLinkedQueue<Subscription> subscriptions = new ConcurrentLinkedQueue<Subscription>();

src/test/java/nz/co/fortytwo/signalk/processor/FullExportProcessorTest.java

@@ -106,7 +106,7 @@ public void shouldEmitIfMatchesWithIdealPolicy() throws Exception {
 
     @Test
     public void shouldEmitIfMatchesWithInstantPolicy() throws Exception {
-        testScenario("vessels.self.navigation", SignalKConstants.POLICY_INSTANT, 3, self_nav);
+        testScenario("vessels.self.navigation", SignalKConstants.POLICY_INSTANT, 1, self_nav);
     }
 
     @Test

src/test/java/nz/co/fortytwo/signalk/processor/SourceToSourceRefProcessorTest.java

@@ -25,6 +25,7 @@
  */
 package nz.co.fortytwo.signalk.processor;
 
+import static nz.co.fortytwo.signalk.util.SignalKConstants.MSG_TYPE;
 import static nz.co.fortytwo.signalk.util.SignalKConstants.dot;
 import static nz.co.fortytwo.signalk.util.SignalKConstants.env_outside_pressure;
 import static nz.co.fortytwo.signalk.util.SignalKConstants.nav_courseOverGroundMagnetic;
@@ -61,9 +62,11 @@ public void shouldMoveSource() throws Exception {
 
 		model.putAll(TestHelper.getSourcesModel().getFullData());
 		logger.debug("Input:"+model);
+		logger.debug("Input vessels:"+model.getSubMap("vessels"));
 		SourceToSourceRefProcessor processor = new SourceToSourceRefProcessor();
 		Exchange ex = new DefaultExchange(ctx);
 		ex.getIn().setBody(model);
+		ex.getIn().setHeader(MSG_TYPE,"unknown");
 		processor.process(ex);
 		logger.debug("Processed to:"+ex.getIn().getBody());
 		String srcRef = "unknown";

src/test/resources/log4j2.json

@@ -59,11 +59,11 @@
 			},
 			{
 				"name": "nz.co.fortytwo.signalk.handler",
-				"level": "info"
+				"level": "debug"
 			},
 			{
 				"name": "nz.co.fortytwo.signalk.processor",
-				"level": "info"
+				"level": "debug"
 			},
 			{
 				"name": "nz.co.fortytwo.signalk.processor.InputFilterProcessor",