Merge branch 'master' into develop

src/conf.c

@@ -2174,6 +2174,8 @@ static int config__read_file_core(struct mosquitto__config *config, bool reload,
 						return MOSQ_ERR_INVAL;
 					}
 					cur_bridge->restart_timeout = atoi(token);
+					cur_bridge->backoff_base = 0;
+					cur_bridge->backoff_cap = 0;
 					if(cur_bridge->restart_timeout < 1){
 						log__printf(NULL, MOSQ_LOG_NOTICE, "restart_timeout interval too low, using 1 second.");
 						cur_bridge->restart_timeout = 1;

src/database.c

@@ -1223,6 +1223,40 @@ int db__message_release_incoming(struct mosquitto *context, uint16_t mid)
 	}
 }
 
+
+void db__expire_all_messages(struct mosquitto *context)
+{
+	struct mosquitto_client_msg *msg, *tmp;
+
+	DL_FOREACH_SAFE(context->msgs_out.inflight, msg, tmp){
+		if(msg->base_msg->message_expiry_time && db.now_real_s > msg->base_msg->message_expiry_time){
+			if(msg->qos > 0){
+				util__increment_send_quota(context);
+			}
+			db__message_remove_inflight(context, &context->msgs_out, msg);
+		}
+	}
+	DL_FOREACH_SAFE(context->msgs_out.queued, msg, tmp){
+		if(msg->base_msg->message_expiry_time && db.now_real_s > msg->base_msg->message_expiry_time){
+			db__message_remove_queued(context, &context->msgs_out, msg);
+		}
+	}
+	DL_FOREACH_SAFE(context->msgs_in.inflight, msg, tmp){
+		if(msg->base_msg->message_expiry_time && db.now_real_s > msg->base_msg->message_expiry_time){
+			if(msg->qos > 0){
+				util__increment_receive_quota(context);
+			}
+			db__message_remove_inflight(context, &context->msgs_in, msg);
+		}
+	}
+	DL_FOREACH_SAFE(context->msgs_in.queued, msg, tmp){
+		if(msg->base_msg->message_expiry_time && db.now_real_s > msg->base_msg->message_expiry_time){
+			db__message_remove_queued(context, &context->msgs_in, msg);
+		}
+	}
+}
+
+
 static int db__message_write_inflight_out_single(struct mosquitto *context, struct mosquitto_client_msg *msg)
 {
 	mosquitto_property *base_msg_props = NULL;
@@ -1392,7 +1426,7 @@ int db__message_write_queued_in(struct mosquitto *context)
 	}
 
 	DL_FOREACH_SAFE(context->msgs_in.queued, tail, tmp){
-		if(context->msgs_out.inflight_maximum != 0 && context->msgs_in.inflight_quota == 0){
+		if(context->msgs_in.inflight_maximum != 0 && context->msgs_in.inflight_quota == 0){
 			break;
 		}
 

src/handle_connack.c

@@ -175,7 +175,7 @@ int handle__connack(struct mosquitto *context)
 					log__printf(NULL, MOSQ_LOG_ERR, "Connection Refused: broker unavailable");
 					return MOSQ_ERR_CONN_LOST;
 				case CONNACK_REFUSED_BAD_USERNAME_PASSWORD:
-					log__printf(NULL, MOSQ_LOG_ERR, "Connection Refused: broker unavailable");
+					log__printf(NULL, MOSQ_LOG_ERR, "Connection Refused: bad user name or password");
 					return MOSQ_ERR_CONN_LOST;
 				case CONNACK_REFUSED_NOT_AUTHORIZED:
 					log__printf(NULL, MOSQ_LOG_ERR, "Connection Refused: not authorised");

src/handle_connect.c

@@ -334,6 +334,7 @@ int connect__on_authorised(struct mosquitto *context, void *auth_data_out, uint1
 	rc = send__connack(context, connect_ack, CONNACK_ACCEPTED, connack_props);
 	mosquitto_property_free_all(&connack_props);
 	if(rc) return rc;
+	db__expire_all_messages(context);
 	rc = db__message_write_queued_out(context);
 	if(rc) return rc;
 	rc = db__message_write_inflight_out_all(context);
@@ -600,9 +601,6 @@ int handle__connect(struct mosquitto *context)
 		rc = MOSQ_ERR_PROTOCOL;
 		goto handle_connect_error;
 	}
-	if(context->in_packet.command != CMD_CONNECT){
-		return MOSQ_ERR_MALFORMED_PACKET;
-	}
 
 	/* Read protocol name as length then bytes rather than with read_string
 	 * because the length is fixed and we can check that. Removes the need
@@ -681,6 +679,9 @@ int handle__connect(struct mosquitto *context)
 		rc = MOSQ_ERR_PROTOCOL;
 		goto handle_connect_error;
 	}
+	if((protocol_version&0x7F) != PROTOCOL_VERSION_v31 && context->in_packet.command != CMD_CONNECT){
+		return MOSQ_ERR_MALFORMED_PACKET;
+	}
 
 	if(packet__read_byte(&context->in_packet, &connect_flags)){
 		rc = MOSQ_ERR_PROTOCOL;

src/mosquitto_broker_internal.h

@@ -753,6 +753,7 @@ int db__message_write_queued_in(struct mosquitto *context);
 void db__msg_add_to_inflight_stats(struct mosquitto_msg_data *msg_data, struct mosquitto_client_msg *msg);
 void db__msg_add_to_queued_stats(struct mosquitto_msg_data *msg_data, struct mosquitto_client_msg *msg);
 uint64_t db__new_msg_id(void);
+void db__expire_all_messages(struct mosquitto *context);
 
 /* ============================================================
  * Subscription functions
@@ -933,6 +934,7 @@ void unpwd__free_item(struct mosquitto__unpwd **unpwd, struct mosquitto__unpwd *
  * Session expiry
  * ============================================================ */
 int session_expiry__add(struct mosquitto *context);
+int session_expiry__add_from_persistence(struct mosquitto *context, time_t expiry_time);
 void session_expiry__remove(struct mosquitto *context);
 void session_expiry__remove_all(void);
 void session_expiry__check(void);

src/net.c

@@ -19,15 +19,16 @@ Contributors:
 #include "config.h"
 
 #ifndef WIN32
-#include <netdb.h>
-#include <unistd.h>
-#include <arpa/inet.h>
-#include <sys/socket.h>
-#include <netinet/tcp.h>
-#include <ifaddrs.h>
+#  include <arpa/inet.h>
+#  include <ifaddrs.h>
+#  include <netdb.h>
+#  include <netinet/tcp.h>
+#  include <strings.h>
+#  include <sys/socket.h>
+#  include <unistd.h>
 #else
-#include <winsock2.h>
-#include <ws2tcpip.h>
+#  include <winsock2.h>
+#  include <ws2tcpip.h>
 #endif
 
 #include <assert.h>
@@ -36,7 +37,7 @@ Contributors:
 #include <stdio.h>
 #include <string.h>
 #ifdef WITH_WRAP
-#include <tcpd.h>
+#  include <tcpd.h>
 #endif
 
 #ifdef HAVE_NETINET_IN_H
@@ -49,7 +50,7 @@ Contributors:
 #endif
 
 #ifdef __QNX__
-#include <net/netbyte.h>
+#  include <net/netbyte.h>
 #endif
 
 #include "mosquitto_broker_internal.h"
@@ -59,8 +60,8 @@ Contributors:
 #include "util_mosq.h"
 
 #ifdef WITH_TLS
-#include "tls_mosq.h"
-#include <openssl/err.h>
+#  include "tls_mosq.h"
+#  include <openssl/err.h>
 static int tls_ex_index_context = -1;
 static int tls_ex_index_listener = -1;
 #endif
@@ -611,7 +612,7 @@ int net__tls_load_verify(struct mosquitto__listener *listener)
 #ifdef WITH_TLS
 	int rc;
 
-#if OPENSSL_VERSION_NUMBER < 0x30000000L
+#  if OPENSSL_VERSION_NUMBER < 0x30000000L
 	if(listener->cafile || listener->capath){
 		rc = SSL_CTX_load_verify_locations(listener->ssl_ctx, listener->cafile, listener->capath);
 		if(rc == 0){
@@ -624,7 +625,7 @@ int net__tls_load_verify(struct mosquitto__listener *listener)
 			}
 		}
 	}
-#else
+#  else
 	if(listener->cafile){
 		rc = SSL_CTX_load_verify_file(listener->ssl_ctx, listener->cafile);
 		if(rc == 0){
@@ -641,11 +642,13 @@ int net__tls_load_verify(struct mosquitto__listener *listener)
 			return MOSQ_ERR_TLS;
 		}
 	}
-#endif
+#  endif
 
+#  if !defined(OPENSSL_NO_ENGINE)
 	if(net__load_engine(listener)){
 		return MOSQ_ERR_TLS;
 	}
+#  endif
 #endif
 	return net__load_certificates(listener);
 }

src/persist_read.c

@@ -208,7 +208,7 @@ static int persist__client_chunk_restore(FILE *db_fptr)
 				}
 			}
 		}
-		/* FIXME - we should expire clients here if they have exceeded their time */
+		session_expiry__add_from_persistence(context, chunk.F.session_expiry_time);
 	}else{
 		rc = 1;
 	}

src/persist_write.c

@@ -164,8 +164,17 @@ static int persist__client_save(FILE *db_fptr)
 	memset(&chunk, 0, sizeof(struct P_client));
 
 	HASH_ITER(hh_id, db.contexts_by_id, context, ctxt_tmp){
-		if(context && context->clean_start == false){
+		if(context && (context->clean_start == false
+#ifdef WITH_BRIDGE
+				|| (context->bridge && context->bridge->clean_start_local == false)
+#endif
+				)){
 			chunk.F.session_expiry_time = context->session_expiry_time;
+			if(context->session_expiry_interval != 0 && context->session_expiry_interval != UINT32_MAX && context->session_expiry_time == 0){
+				chunk.F.session_expiry_time = context->session_expiry_interval + db.now_real_s;
+			}else{
+				chunk.F.session_expiry_time = context->session_expiry_time;
+			}
 			chunk.F.session_expiry_interval = context->session_expiry_interval;
 			chunk.F.last_mid = context->last_mid;
 			chunk.F.id_len = (uint16_t)strlen(context->id);

src/session_expiry.c

@@ -84,6 +84,31 @@ int session_expiry__add(struct mosquitto *context)
 }
 
 
+int session_expiry__add_from_persistence(struct mosquitto *context, time_t expiry_time)
+{
+	struct session_expiry_list *item;
+
+	if(db.config->persistent_client_expiration == 0){
+		if(context->session_expiry_interval == UINT32_MAX){
+			/* There isn't a global expiry set, and the client has asked to
+			 * never expire, so we don't add it to the list. */
+			return MOSQ_ERR_SUCCESS;
+		}
+	}
+
+	item = mosquitto__calloc(1, sizeof(struct session_expiry_list));
+	if(!item) return MOSQ_ERR_NOMEM;
+
+	item->context = context;
+	item->context->session_expiry_time = expiry_time;
+	context->expiry_list_item = item;
+
+	DL_INSERT_INORDER(expiry_list, item, session_expiry__cmp);
+
+	return MOSQ_ERR_SUCCESS;
+}
+
+
 void session_expiry__remove(struct mosquitto *context)
 {
 	if(context->expiry_list_item){

src/websockets.c

@@ -136,6 +136,7 @@ static int callback_mqtt(
 	uint8_t *buf;
 	int rc;
 	uint8_t byte;
+	char ip_addr_buff[1024];
 
 	switch (reason) {
 		case LWS_CALLBACK_ESTABLISHED:
@@ -160,7 +161,12 @@ static int callback_mqtt(
 			}else{
 				return -1;
 			}
-			easy_address(lws_get_socket_fd(wsi), mosq);
+
+			if (lws_hdr_copy(wsi, ip_addr_buff, sizeof(ip_addr_buff), WSI_TOKEN_X_FORWARDED_FOR) > 0) {
+				mosq->address = mosquitto__strdup(ip_addr_buff);
+			} else {
+				easy_address(lws_get_socket_fd(wsi), mosq);
+			}
 			if(!mosq->address){
 				/* getpeername and inet_ntop failed and not a bridge */
 				mosquitto__FREE(mosq);
@@ -246,7 +252,7 @@ static int callback_mqtt(
 
 #ifdef WITH_SYS_TREE
 				g_msgs_sent++;
-				if(((packet->command)&0xF6) == CMD_PUBLISH){
+				if(((packet->command)&0xF0) == CMD_PUBLISH){
 					g_pub_msgs_sent++;
 				}
 #endif
@@ -331,7 +337,7 @@ static int callback_mqtt(
 
 #ifdef WITH_SYS_TREE
 				G_MSGS_RECEIVED_INC(1);
-				if(((mosq->in_packet.command)&0xF5) == CMD_PUBLISH){
+				if(((mosq->in_packet.command)&0xF0) == CMD_PUBLISH){
 					G_PUB_MSGS_RECEIVED_INC(1);
 				}
 #endif