Add hop limit to json and proto formats.

adsbus/Makefile

@@ -27,7 +27,7 @@ clean:
 adsb.pb-c.c: ../proto/adsb.proto
 	protoc-c --c_out=./ --proto_path=$(dir $<) $<
 
-adsbus: adsbus.o $(OBJ_TRANSPORT) $(OBJ_FLOW) $(OBJ_PROTOCOL) $(OBJ_UTIL) $(OBJ_PROTO)
+adsbus: adsbus.o $(OBJ_PROTO) $(OBJ_TRANSPORT) $(OBJ_FLOW) $(OBJ_PROTOCOL) $(OBJ_UTIL)
 	$(COMP) $(LDFLAGS) -o adsbus adsbus.o $(OBJ_TRANSPORT) $(OBJ_FLOW) $(OBJ_PROTOCOL) $(OBJ_UTIL) $(OBJ_PROTO) $(LIBS)
 
 afl-fuzz:

adsbus/adsb.pb-c.c

@@ -282,7 +282,7 @@ const ProtobufCMessageDescriptor adsb_header__descriptor =
   (ProtobufCMessageInit) adsb_header__init,
   NULL,NULL,NULL    /* reserved[123] */
 };
-static const ProtobufCFieldDescriptor adsb_packet__field_descriptors[4] =
+static const ProtobufCFieldDescriptor adsb_packet__field_descriptors[5] =
 {
   {
     "source_id",
@@ -297,8 +297,20 @@ static const ProtobufCFieldDescriptor adsb_packet__field_descriptors[4] =
     0,NULL,NULL    /* reserved1,reserved2, etc */
   },
   {
-    "mlat_timestamp",
+    "hops",
     2,
+    PROTOBUF_C_LABEL_REQUIRED,
+    PROTOBUF_C_TYPE_UINT32,
+    0,   /* quantifier_offset */
+    offsetof(AdsbPacket, hops),
+    NULL,
+    NULL,
+    0,             /* flags */
+    0,NULL,NULL    /* reserved1,reserved2, etc */
+  },
+  {
+    "mlat_timestamp",
+    3,
     PROTOBUF_C_LABEL_OPTIONAL,
     PROTOBUF_C_TYPE_FIXED64,
     offsetof(AdsbPacket, has_mlat_timestamp),
@@ -310,7 +322,7 @@ static const ProtobufCFieldDescriptor adsb_packet__field_descriptors[4] =
   },
   {
     "rssi",
-    3,
+    4,
     PROTOBUF_C_LABEL_OPTIONAL,
     PROTOBUF_C_TYPE_FIXED32,
     offsetof(AdsbPacket, has_rssi),
@@ -322,7 +334,7 @@ static const ProtobufCFieldDescriptor adsb_packet__field_descriptors[4] =
   },
   {
     "payload",
-    4,
+    5,
     PROTOBUF_C_LABEL_REQUIRED,
     PROTOBUF_C_TYPE_BYTES,
     0,   /* quantifier_offset */
@@ -334,15 +346,16 @@ static const ProtobufCFieldDescriptor adsb_packet__field_descriptors[4] =
   },
 };
 static const unsigned adsb_packet__field_indices_by_name[] = {
-  1,   /* field[1] = mlat_timestamp */
+  1,   /* field[1] = hops */
-  3,   /* field[3] = payload */
+  2,   /* field[2] = mlat_timestamp */
-  2,   /* field[2] = rssi */
+  4,   /* field[4] = payload */
+  3,   /* field[3] = rssi */
   0,   /* field[0] = source_id */
 };
 static const ProtobufCIntRange adsb_packet__number_ranges[1 + 1] =
 {
   { 1, 0 },
-  { 0, 4 }
+  { 0, 5 }
 };
 const ProtobufCMessageDescriptor adsb_packet__descriptor =
 {
@@ -352,7 +365,7 @@ const ProtobufCMessageDescriptor adsb_packet__descriptor =
   "AdsbPacket",
   "",
   sizeof(AdsbPacket),
-  4,
+  5,
   adsb_packet__field_descriptors,
   adsb_packet__field_indices_by_name,
   1,  adsb_packet__number_ranges,

adsbus/adsb.pb-c.h

@@ -45,6 +45,7 @@ struct  _AdsbPacket
 {
   ProtobufCMessage base;
   char *source_id;
+  uint32_t hops;
   protobuf_c_boolean has_mlat_timestamp;
   uint64_t mlat_timestamp;
   protobuf_c_boolean has_rssi;
@@ -53,7 +54,7 @@ struct  _AdsbPacket
 };
 #define ADSB_PACKET__INIT \
  { PROTOBUF_C_MESSAGE_INIT (&adsb_packet__descriptor) \
-    , NULL, 0,0, 0,0, {0,NULL} }
+    , NULL, 0, 0,0, 0,0, {0,NULL} }
 
 
 struct  _Adsb

adsbus/adsbus.c

@@ -183,6 +183,7 @@ int main(int argc, char *argv[]) {
 	wakeup_init();
 	peer_init();
 
+	receive_init();
 	send_init();
 
 	beast_init();

adsbus/json.c

@@ -34,6 +34,7 @@ static void json_serialize_to_buf(json_t *obj, struct buf *buf) {
 }
 
 static void json_add_common(struct packet *packet, json_t *obj) {
+	json_object_set_new(obj, "hops", json_integer(packet->hops));
 	json_object_set_new(obj, "type", json_string(packet_type_names[packet->type]));
 	json_object_set_new(obj, "source_id", json_string((const char *) packet->source_id));
 	if (packet->mlat_timestamp) {
@@ -78,10 +79,6 @@ static bool json_parse_header(json_t *in, struct packet *packet, struct json_par
 		return false;
 	}
 
-	state->mlat_timestamp_mhz = (uint16_t) mlat_timestamp_mhz;
-	state->mlat_timestamp_max = (uint64_t) mlat_timestamp_max;
-	state->rssi_max = (uint32_t) rssi_max;
-
 	if (!strcmp(json_server_id, (const char *) server_id)) {
 		fprintf(stderr, "R %s: Attempt to receive json data from our own server ID (%s); loop!\n", packet->source_id, server_id);
 		return false;
@@ -89,6 +86,10 @@ static bool json_parse_header(json_t *in, struct packet *packet, struct json_par
 
 	fprintf(stderr, "R %s: Connected to server ID: %s\n", packet->source_id, json_server_id);
 
+	state->mlat_timestamp_mhz = (uint16_t) mlat_timestamp_mhz;
+	state->mlat_timestamp_max = (uint64_t) mlat_timestamp_max;
+	state->rssi_max = (uint32_t) rssi_max;
+
 	state->have_header = true;
 	packet->type = PACKET_TYPE_NONE;
 	return true;
@@ -99,7 +100,12 @@ static bool json_parse_common(json_t *in, struct packet *packet, struct json_par
 		return false;
 	}
 
-	if (json_unpack(in, "{s:s}", "source_id", &packet->source_id)) {
+	json_int_t hops;
+
+	if (json_unpack(
+			in, "{s:s, s:I}",
+			"source_id", &packet->source_id,
+			"hops", &hops)) {
 		return false;
 	}
 
@@ -107,6 +113,11 @@ static bool json_parse_common(json_t *in, struct packet *packet, struct json_par
 		return false;
 	}
 
+	if (hops < 0 || hops > UINT32_MAX) {
+		return false;
+	}
+	packet->hops = (uint16_t) hops;
+
 	json_t *mlat_timestamp = json_object_get(in, "mlat_timestamp");
 	if (mlat_timestamp && json_is_integer(mlat_timestamp)) {
 		json_int_t val = json_integer_value(mlat_timestamp);

adsbus/packet.h

@@ -16,6 +16,7 @@ struct packet {
 		PACKET_TYPE_MODE_S_LONG,
 	} type;
 #define NUM_TYPES 4
+	uint32_t hops;
 	uint8_t payload[PACKET_DATA_LEN_MAX];
 	uint64_t mlat_timestamp;
 	uint32_t rssi;

adsbus/proto.c

@@ -43,6 +43,7 @@ static void proto_wrap_to_buf(Adsb *msg, struct buf *buf) {
 
 static void proto_serialize_packet(struct packet *packet, AdsbPacket *out, size_t len) {
 	out->source_id = (char *) packet->source_id;
+	out->hops = packet->hops;
 	if (packet->mlat_timestamp) {
 		out->mlat_timestamp = packet->mlat_timestamp;
 		out->has_mlat_timestamp = true;
@@ -165,7 +166,9 @@ static bool proto_parse_packet(AdsbPacket *in, struct packet *packet, struct pro
 	if (!packet_validate_id((const uint8_t *) in->source_id)) {
 		return false;
 	}
+
 	packet->source_id = (uint8_t *) in->source_id;
+	packet->hops = (uint16_t) in->hops;
 	memcpy(packet->payload, in->payload.data, len);
 
 	if (in->has_mlat_timestamp) {

adsbus/receive.c

@@ -76,6 +76,8 @@ static struct parser {
 };
 #define NUM_PARSERS (sizeof(parsers) / sizeof(*parsers))
 
+static uint32_t receive_max_hops = 10;
+
 static bool receive_parse_wrapper(struct receive *receive, struct packet *packet) {
 	return receive->parser(&receive->buf, packet, receive->parser_state);
 }
@@ -84,6 +86,10 @@ static bool receive_autodetect_parse(struct receive *receive, struct packet *pac
 	struct buf *buf = &receive->buf;
 	void *state = receive->parser_state;
 
+	// We don't trust parsers not to scribble over the packet.
+	struct packet orig_packet;
+	memcpy(&orig_packet, packet, sizeof(orig_packet));
+
 	for (size_t i = 0; i < NUM_PARSERS; i++) {
 		if (parsers[i].parse(buf, packet, state)) {
 			fprintf(stderr, "R %s: Detected input format %s\n", receive->id, parsers[i].name);
@@ -91,6 +97,9 @@ static bool receive_autodetect_parse(struct receive *receive, struct packet *pac
 			receive->parser = parsers[i].parse;
 			return true;
 		}
+		if (i < NUM_PARSERS - 1) {
+			memcpy(packet, &orig_packet, sizeof(*packet));
+		}
 	}
 	return false;
 }
@@ -124,6 +133,10 @@ static void receive_read(struct peer *peer) {
 		if (packet.type == PACKET_TYPE_NONE) {
 			continue;
 		}
+		if (++packet.hops > receive_max_hops) {
+			fprintf(stderr, "R %s: Packet exceeded hop limit (%u > %u); dropping. You may have a loop in your configuration.\n", receive->id, packet.hops, receive_max_hops);
+			continue;
+		}
 		send_write(&packet);
 	}
 
@@ -156,6 +169,18 @@ static void receive_new(int fd, void __attribute__((unused)) *passthrough, struc
 	fprintf(stderr, "R %s: New receive connection\n", receive->id);
 }
 
+void receive_init() {
+	char *max_hops = getenv("ADSBUS_MAX_HOPS");
+	if (max_hops) {
+		char *end_ptr;
+		unsigned long max_hops_ul = strtoul(max_hops, &end_ptr, 10);
+		assert(max_hops[0] != '\0');
+		assert(end_ptr[0] == '\0');
+		assert(max_hops_ul <= UINT32_MAX);
+		receive_max_hops = (uint32_t) max_hops_ul;
+	}
+}
+
 void receive_cleanup() {
 	struct receive *iter, *next;
 	list_for_each_entry_safe(iter, next, &receive_head, receive_list) {

adsbus/receive.h

@@ -4,6 +4,7 @@
 
 struct flow;
 
+void receive_init(void);
 void receive_cleanup(void);
 void receive_print_usage(void);
 extern struct flow *receive_flow;

proto/adsb.proto

@@ -27,22 +27,25 @@ message AdsbPacket {
 	// 36 character limit
 	required string source_id = 1;
 
+	// Number of routing hops since source, when carried by protocols with a hop count.
+	required uint32 hops = 2;
+
 	// Value of the MLAT counter when this packet arrived at the recorder
 	// Range [0, mlat_timestamp_max]
 	// Units of 1 / (mlat_timestamp_mhz * 10^6) Hz
-	optional fixed64 mlat_timestamp = 2;
+	optional fixed64 mlat_timestamp = 3;
 
 	// RSSI of the received packet at the recorder
 	// Range [0, rssi_max]
 	// Units unspecified
-	optional fixed32 rssi = 3;
+	optional fixed32 rssi = 4;
 
 	// Binary packet payload.
 	// Length:
 	//   mode_ac: 2 bytes
 	//   mode_s_short: 7 bytes
 	//   mode_s_long: 14 bytes
-	required bytes payload = 4;
+	required bytes payload = 5;
 }
 
 message Adsb {