diff --git a/CMakeLists.txt b/CMakeLists.txt index d326ea2b..92604490 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -11,7 +11,7 @@ project(mosquitto) cmake_minimum_required(VERSION 2.8) # Only for version 3 and up. cmake_policy(SET CMP0042 NEW) -set (VERSION 1.6.10) +set (VERSION 1.6.11) add_definitions (-DCMAKE -DVERSION=\"${VERSION}\") diff --git a/ChangeLog.txt b/ChangeLog.txt index a40e202b..1630e251 100644 --- a/ChangeLog.txt +++ b/ChangeLog.txt @@ -1,3 +1,45 @@ +1.6.11 - 2020-08-11 +=================== + +Security: +- On Windows the Mosquitto service was being installed without appropriate + path quoting, this has been fixed. + +Broker: +- Fix usage message only mentioning v3.1.1. Closes #1713. +- Fix broker refusing to start if only websockets listeners were defined. + Closes #1740. +- Change systemd unit files to create /var/log/mosquitto before starting. + Closes #821. +- Don't quit with an error if opening the log file isn't possible. + Closes #821. +- Fix bridge topic remapping when using "" as the topic. Closes #1749. +- Fix messages being queued for disconnected bridges when clean start was + set to true. Closes #1729. +- Fix `autosave_interval` not being triggered by messages being delivered. + Closes #1726. +- Fix websockets clients sometimes not being disconnected promptly. + Closes #1718. +- Fix "slow" file based logging by switching to line based buffering. + Closes #1689. Closes #1741. +- Log protocol error message where appropriate from a bad UNSUBSCRIBE, rather + than the generic "socket error". +- Don't try to start DLT logging if DLT unavailable, to avoid a long delay + when shutting down the broker. Closes #1735. +- Fix potential memory leaks. Closes #1773. Closes #1774. +- Fix clients not receiving messages after a previous client with the same + client ID and positive will delay interval quit. Closes #1752. +- Fix overly broad HAVE_PTHREAD_CANCEL compile guard. Closes #1547. + +Client library: +- Improved documentation around connect callback return codes. Close #1730. +- Fix `mosquitto_publish*()` no longer returning `MOSQ_ERR_NO_CONN` when not + connected. Closes #1725. +- `mosquitto_loop_start()` now sets a thread name on Linux, FreeBSD, NetBSD, + and OpenBSD. Closes #1777. +- Fix `mosquitto_loop_stop()` not stopping on Windows. Closes #1748. Closes #117. + + 1.6.10 - 2020-05-25 =================== diff --git a/client/pub_shared.c b/client/pub_shared.c index ae4cb277..015c5eb1 100644 --- a/client/pub_shared.c +++ b/client/pub_shared.c @@ -71,7 +71,7 @@ int load_stdin(void) pos += rlen; } if(pos > MQTT_MAX_PAYLOAD){ - err_printf(&cfg, "Error: Message length must be less that %u bytes.\n\n", MQTT_MAX_PAYLOAD); + err_printf(&cfg, "Error: Message length must be less than %u bytes.\n\n", MQTT_MAX_PAYLOAD); free(cfg.message); return 1; } @@ -100,13 +100,9 @@ int load_file(const char *filename) fseek(fptr, 0, SEEK_END); flen = ftell(fptr); if(flen > MQTT_MAX_PAYLOAD){ - err_printf(&cfg, "Error: Message length must be less that %u bytes.\n\n", MQTT_MAX_PAYLOAD); - free(cfg.message); - return 1; - } - if(flen > 268435455){ fclose(fptr); - err_printf(&cfg, "Error: File \"%s\" is too large (>268,435,455 bytes).\n", filename); + err_printf(&cfg, "Error: File must be less than %u bytes.\n\n", MQTT_MAX_PAYLOAD); + free(cfg.message); return 1; }else if(flen == 0){ fclose(fptr); diff --git a/config.mk b/config.mk index 89e0c319..75c8b35f 100644 --- a/config.mk +++ b/config.mk @@ -109,7 +109,7 @@ WITH_COVERAGE:=no # Also bump lib/mosquitto.h, CMakeLists.txt, # installer/mosquitto.nsi, installer/mosquitto64.nsi -VERSION=1.6.10 +VERSION=1.6.11 # Client library SO version. Bump if incompatible API/ABI changes are made. SOVERSION=1 diff --git a/docker/1.5/docker-entrypoint.sh b/docker/1.5/docker-entrypoint.sh index b381ac57..85099f54 100755 --- a/docker/1.5/docker-entrypoint.sh +++ b/docker/1.5/docker-entrypoint.sh @@ -1,4 +1,15 @@ #!/bin/ash set -e +docker_set_permissions() { + local user; user="$(id -u)" + + if [ "$user" = '0' ]; then + chown -R mosquitto:mosquitto /mosquitto + fi +} + + +docker_set_permissions() + exec "$@" diff --git a/docker/1.6/docker-entrypoint.sh b/docker/1.6/docker-entrypoint.sh index b381ac57..85099f54 100755 --- a/docker/1.6/docker-entrypoint.sh +++ b/docker/1.6/docker-entrypoint.sh @@ -1,4 +1,15 @@ #!/bin/ash set -e +docker_set_permissions() { + local user; user="$(id -u)" + + if [ "$user" = '0' ]; then + chown -R mosquitto:mosquitto /mosquitto + fi +} + + +docker_set_permissions() + exec "$@" diff --git a/docker/generic/Dockerfile b/docker/generic/Dockerfile index 305f3f0f..cc467946 100644 --- a/docker/generic/Dockerfile +++ b/docker/generic/Dockerfile @@ -6,7 +6,7 @@ RUN test -n "${VERSION}" RUN apk --no-cache add \ build-base \ - libressl-dev \ + openssl-dev \ c-ares-dev \ curl \ util-linux-dev \ @@ -48,8 +48,7 @@ LABEL maintainer="Jonathan Hanson " \ RUN apk --no-cache add \ busybox \ ca-certificates \ - libcrypto1.0 \ - libssl1.0 \ + openssl \ libuuid \ libwebsockets \ musl diff --git a/docker/generic/docker-entrypoint.sh b/docker/generic/docker-entrypoint.sh index b381ac57..85099f54 100755 --- a/docker/generic/docker-entrypoint.sh +++ b/docker/generic/docker-entrypoint.sh @@ -1,4 +1,15 @@ #!/bin/ash set -e +docker_set_permissions() { + local user; user="$(id -u)" + + if [ "$user" = '0' ]; then + chown -R mosquitto:mosquitto /mosquitto + fi +} + + +docker_set_permissions() + exec "$@" diff --git a/docker/local/docker-entrypoint.sh b/docker/local/docker-entrypoint.sh index b381ac57..85099f54 100755 --- a/docker/local/docker-entrypoint.sh +++ b/docker/local/docker-entrypoint.sh @@ -1,4 +1,15 @@ #!/bin/ash set -e +docker_set_permissions() { + local user; user="$(id -u)" + + if [ "$user" = '0' ]; then + chown -R mosquitto:mosquitto /mosquitto + fi +} + + +docker_set_permissions() + exec "$@" diff --git a/examples/temperature_conversion/readme.txt b/examples/temperature_conversion/readme.txt index a706a7cc..de9d9a19 100644 --- a/examples/temperature_conversion/readme.txt +++ b/examples/temperature_conversion/readme.txt @@ -2,5 +2,5 @@ This is a simple example of the C++ library mosquittopp. It is a client that subscribes to the topic temperature/celsius which should have temperature data in text form being published to it. It reads this data as -a Celsius temperature, converts to Farenheit and republishes on -temperature/farenheit. +a Celsius temperature, converts to Fahrenheit and republishes on +temperature/fahrenheit. diff --git a/examples/temperature_conversion/temperature_conversion.cpp b/examples/temperature_conversion/temperature_conversion.cpp index 1fa6f55e..7ecf0d80 100644 --- a/examples/temperature_conversion/temperature_conversion.cpp +++ b/examples/temperature_conversion/temperature_conversion.cpp @@ -28,7 +28,7 @@ void mqtt_tempconv::on_connect(int rc) void mqtt_tempconv::on_message(const struct mosquitto_message *message) { - double temp_celsius, temp_farenheit; + double temp_celsius, temp_fahrenheit; char buf[51]; if(!strcmp(message->topic, "temperature/celsius")){ @@ -36,9 +36,9 @@ void mqtt_tempconv::on_message(const struct mosquitto_message *message) /* Copy N-1 bytes to ensure always 0 terminated. */ memcpy(buf, message->payload, 50*sizeof(char)); temp_celsius = atof(buf); - temp_farenheit = temp_celsius*9.0/5.0 + 32.0; - snprintf(buf, 50, "%f", temp_farenheit); - publish(NULL, "temperature/farenheit", strlen(buf), buf); + temp_fahrenheit = temp_celsius*9.0/5.0 + 32.0; + snprintf(buf, 50, "%f", temp_fahrenheit); + publish(NULL, "temperature/fahrenheit", strlen(buf), buf); } } diff --git a/installer/mosquitto.nsi b/installer/mosquitto.nsi index 4df8ac3d..6cd04601 100644 --- a/installer/mosquitto.nsi +++ b/installer/mosquitto.nsi @@ -9,7 +9,7 @@ !define env_hklm 'HKLM "SYSTEM\CurrentControlSet\Control\Session Manager\Environment"' Name "Eclipse Mosquitto" -!define VERSION 1.6.10 +!define VERSION 1.6.11 OutFile "mosquitto-${VERSION}-install-windows-x86.exe" InstallDir "$PROGRAMFILES\mosquitto" diff --git a/installer/mosquitto64.nsi b/installer/mosquitto64.nsi index cd4ddc8f..dc54654b 100644 --- a/installer/mosquitto64.nsi +++ b/installer/mosquitto64.nsi @@ -9,7 +9,7 @@ !define env_hklm 'HKLM "SYSTEM\CurrentControlSet\Control\Session Manager\Environment"' Name "Eclipse Mosquitto" -!define VERSION 1.6.10 +!define VERSION 1.6.11 OutFile "mosquitto-${VERSION}-install-windows-x64.exe" !include "x64.nsh" diff --git a/lib/actions.c b/lib/actions.c index 2b6dc596..d3914acf 100644 --- a/lib/actions.c +++ b/lib/actions.c @@ -151,9 +151,9 @@ int mosquitto_publish_v5(struct mosquitto *mosq, int *mid, const char *topic, in pthread_mutex_lock(&mosq->msgs_out.mutex); message->state = mosq_ms_invalid; - message__queue(mosq, message, mosq_md_out); + rc = message__queue(mosq, message, mosq_md_out); pthread_mutex_unlock(&mosq->msgs_out.mutex); - return MOSQ_ERR_SUCCESS; + return rc; } } diff --git a/lib/dummypthread.h b/lib/dummypthread.h index 4207f3d6..611a1574 100644 --- a/lib/dummypthread.h +++ b/lib/dummypthread.h @@ -4,6 +4,7 @@ #define pthread_create(A, B, C, D) #define pthread_join(A, B) #define pthread_cancel(A) +#define pthread_testcancel() #define pthread_mutex_init(A, B) #define pthread_mutex_destroy(A) diff --git a/lib/loop.c b/lib/loop.c index 048b8211..a3773b0a 100644 --- a/lib/loop.c +++ b/lib/loop.c @@ -256,6 +256,9 @@ int mosquitto_loop_forever(struct mosquitto *mosq, int timeout, int max_packets) while(run){ do{ +#ifdef HAVE_PTHREAD_CANCEL + pthread_testcancel(); +#endif rc = mosquitto_loop(mosq, timeout, max_packets); }while(run && rc == MOSQ_ERR_SUCCESS); /* Quit after fatal errors. */ @@ -280,6 +283,9 @@ int mosquitto_loop_forever(struct mosquitto *mosq, int timeout, int max_packets) return rc; } do{ +#ifdef HAVE_PTHREAD_CANCEL + pthread_testcancel(); +#endif rc = MOSQ_ERR_SUCCESS; state = mosquitto__get_state(mosq); if(state == mosq_cs_disconnecting || state == mosq_cs_disconnected){ diff --git a/lib/mosquitto.c b/lib/mosquitto.c index f1f200fd..25a4441b 100644 --- a/lib/mosquitto.c +++ b/lib/mosquitto.c @@ -151,10 +151,6 @@ int mosquitto_reinitialise(struct mosquitto *mosq, const char *id, bool clean_st } mosq->protocol = mosq_p_mqtt311; mosq->sock = INVALID_SOCKET; - if(net__socketpair(&mosq->sockpairR, &mosq->sockpairW)){ - log__printf(mosq, MOSQ_LOG_WARNING, - "Warning: Unable to open socket pair, outgoing publish commands may be delayed."); - } mosq->keepalive = 60; mosq->clean_start = clean_start; if(id){ @@ -213,6 +209,12 @@ int mosquitto_reinitialise(struct mosquitto *mosq, const char *id, bool clean_st pthread_mutex_init(&mosq->mid_mutex, NULL); mosq->thread_id = pthread_self(); #endif + /* This must be after pthread_mutex_init(), otherwise the log mutex may be + * used before being initialised. */ + if(net__socketpair(&mosq->sockpairR, &mosq->sockpairW)){ + log__printf(mosq, MOSQ_LOG_WARNING, + "Warning: Unable to open socket pair, outgoing publish commands may be delayed."); + } return MOSQ_ERR_SUCCESS; } diff --git a/lib/mosquitto.h b/lib/mosquitto.h index 879184dc..9d3557d6 100644 --- a/lib/mosquitto.h +++ b/lib/mosquitto.h @@ -48,7 +48,7 @@ extern "C" { #define LIBMOSQUITTO_MAJOR 1 #define LIBMOSQUITTO_MINOR 6 -#define LIBMOSQUITTO_REVISION 10 +#define LIBMOSQUITTO_REVISION 11 /* LIBMOSQUITTO_VERSION_NUMBER looks like 1002001 for e.g. version 1.2.1. */ #define LIBMOSQUITTO_VERSION_NUMBER (LIBMOSQUITTO_MAJOR*1000000+LIBMOSQUITTO_MINOR*1000+LIBMOSQUITTO_REVISION) @@ -434,7 +434,11 @@ libmosq_EXPORT int mosquitto_username_pw_set(struct mosquitto *mosq, const char * * Returns: * MOSQ_ERR_SUCCESS - on success. - * MOSQ_ERR_INVAL - if the input parameters were invalid. + * MOSQ_ERR_INVAL - if the input parameters were invalid, which could be any of: + * * mosq == NULL + * * host == NULL + * * port < 0 + * * keepalive < 5 * MOSQ_ERR_ERRNO - if a system call returned an error. The variable errno * contains the error code, even on Windows. * Use strerror_r() where available or FormatMessage() on @@ -500,7 +504,11 @@ libmosq_EXPORT int mosquitto_connect_bind(struct mosquitto *mosq, const char *ho * * Returns: * MOSQ_ERR_SUCCESS - on success. - * MOSQ_ERR_INVAL - if the input parameters were invalid. + * MOSQ_ERR_INVAL - if the input parameters were invalid, which could be any of: + * * mosq == NULL + * * host == NULL + * * port < 0 + * * keepalive < 5 * MOSQ_ERR_ERRNO - if a system call returned an error. The variable errno * contains the error code, even on Windows. * Use strerror_r() where available or FormatMessage() on @@ -570,7 +578,11 @@ libmosq_EXPORT int mosquitto_connect_async(struct mosquitto *mosq, const char *h * * Returns: * MOSQ_ERR_SUCCESS - on success. - * MOSQ_ERR_INVAL - if the input parameters were invalid. + * MOSQ_ERR_INVAL - if the input parameters were invalid, which could be any of: + * * mosq == NULL + * * host == NULL + * * port < 0 + * * keepalive < 5 * MOSQ_ERR_ERRNO - if a system call returned an error. The variable errno * contains the error code, even on Windows. * Use strerror_r() where available or FormatMessage() on @@ -604,7 +616,11 @@ libmosq_EXPORT int mosquitto_connect_bind_async(struct mosquitto *mosq, const ch * * Returns: * MOSQ_ERR_SUCCESS - on success. - * MOSQ_ERR_INVAL - if the input parameters were invalid. + * MOSQ_ERR_INVAL - if the input parameters were invalid, which could be any of: + * * mosq == NULL + * * host == NULL + * * port < 0 + * * keepalive < 5 * MOSQ_ERR_ERRNO - if a system call returned an error. The variable errno * contains the error code, even on Windows. * Use strerror_r() where available or FormatMessage() on @@ -1763,13 +1779,10 @@ libmosq_EXPORT int mosquitto_tls_psk_set(struct mosquitto *mosq, const char *psk * Callback Parameters: * mosq - the mosquitto instance making the callback. * obj - the user data provided in - * rc - the return code of the connection response, one of: - * - * * 0 - success - * * 1 - connection refused (unacceptable protocol version) - * * 2 - connection refused (identifier rejected) - * * 3 - connection refused (broker unavailable) - * * 4-255 - reserved for future use + * rc - the return code of the connection response. The values are defined by + * the MQTT protocol version in use. + * For MQTT v5.0, look at section 3.2.2.2 Connect Reason code: https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html + * For MQTT v3.1.1, look at section 3.2.2.3 Connect Return code: http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/mqtt-v3.1.1.html */ libmosq_EXPORT void mosquitto_connect_callback_set(struct mosquitto *mosq, void (*on_connect)(struct mosquitto *, void *, int)); @@ -1787,14 +1800,11 @@ libmosq_EXPORT void mosquitto_connect_callback_set(struct mosquitto *mosq, void * Callback Parameters: * mosq - the mosquitto instance making the callback. * obj - the user data provided in - * rc - the return code of the connection response, one of: + * rc - the return code of the connection response. The values are defined by + * the MQTT protocol version in use. + * For MQTT v5.0, look at section 3.2.2.2 Connect Reason code: https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html + * For MQTT v3.1.1, look at section 3.2.2.3 Connect Return code: http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/mqtt-v3.1.1.html * flags - the connect flags. - * - * * 0 - success - * * 1 - connection refused (unacceptable protocol version) - * * 2 - connection refused (identifier rejected) - * * 3 - connection refused (broker unavailable) - * * 4-255 - reserved for future use */ libmosq_EXPORT void mosquitto_connect_with_flags_callback_set(struct mosquitto *mosq, void (*on_connect)(struct mosquitto *, void *, int, int)); @@ -1812,12 +1822,10 @@ libmosq_EXPORT void mosquitto_connect_with_flags_callback_set(struct mosquitto * * Callback Parameters: * mosq - the mosquitto instance making the callback. * obj - the user data provided in - * rc - the return code of the connection response, one of: - * * 0 - success - * * 1 - connection refused (unacceptable protocol version) - * * 2 - connection refused (identifier rejected) - * * 3 - connection refused (broker unavailable) - * * 4-255 - reserved for future use + * rc - the return code of the connection response. The values are defined by + * the MQTT protocol version in use. + * For MQTT v5.0, look at section 3.2.2.2 Connect Reason code: https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html + * For MQTT v3.1.1, look at section 3.2.2.3 Connect Return code: http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/mqtt-v3.1.1.html * flags - the connect flags. * props - list of MQTT 5 properties, or NULL * diff --git a/lib/socks_mosq.c b/lib/socks_mosq.c index 51d8fb73..06b4116d 100644 --- a/lib/socks_mosq.c +++ b/lib/socks_mosq.c @@ -27,7 +27,7 @@ Contributors: #else # include #endif -#ifdef __FreeBSD__ +#if defined(__FreeBSD__) || defined(__OpenBSD__) # include # include #endif diff --git a/lib/thread_mosq.c b/lib/thread_mosq.c index d32e26fa..b05c1d2a 100644 --- a/lib/thread_mosq.c +++ b/lib/thread_mosq.c @@ -20,6 +20,12 @@ Contributors: #include #endif +#if defined(__linux__) || defined(__NetBSD__) +# include +#elif defined(__FreeBSD__) || defined(__OpenBSD__) +# include +#endif + #include "mosquitto_internal.h" #include "net_mosq.h" #include "util_mosq.h" @@ -28,11 +34,18 @@ void *mosquitto__thread_main(void *obj); int mosquitto_loop_start(struct mosquitto *mosq) { -#if defined(WITH_THREADING) && defined(HAVE_PTHREAD_CANCEL) +#if defined(WITH_THREADING) if(!mosq || mosq->threaded != mosq_ts_none) return MOSQ_ERR_INVAL; mosq->threaded = mosq_ts_self; if(!pthread_create(&mosq->thread_id, NULL, mosquitto__thread_main, mosq)){ +#if defined(__linux__) + pthread_setname_np(mosq->thread_id, "mosquitto loop"); +#elif defined(__NetBSD__) + pthread_setname_np(mosq->thread_id, "%s", "mosquitto loop"); +#elif defined(__FreeBSD__) || defined(__OpenBSD__) + pthread_set_name_np(mosq->thread_id, "mosquitto loop"); +#endif return MOSQ_ERR_SUCCESS; }else{ return MOSQ_ERR_ERRNO; @@ -44,7 +57,7 @@ int mosquitto_loop_start(struct mosquitto *mosq) int mosquitto_loop_stop(struct mosquitto *mosq, bool force) { -#if defined(WITH_THREADING) && defined(HAVE_PTHREAD_CANCEL) +#if defined(WITH_THREADING) # ifndef WITH_BROKER char sockpair_data = 0; # endif @@ -63,9 +76,11 @@ int mosquitto_loop_stop(struct mosquitto *mosq, bool force) #endif } +#ifdef HAVE_PTHREAD_CANCEL if(force){ pthread_cancel(mosq->thread_id); } +#endif pthread_join(mosq->thread_id, NULL); mosq->thread_id = pthread_self(); mosq->threaded = mosq_ts_none; diff --git a/lib/will_mosq.c b/lib/will_mosq.c index 97fa67d0..59639f4a 100644 --- a/lib/will_mosq.c +++ b/lib/will_mosq.c @@ -120,6 +120,7 @@ int will__clear(struct mosquitto *mosq) mosquitto__free(mosq->will); mosq->will = NULL; + mosq->will_delay_interval = 0; return MOSQ_ERR_SUCCESS; } diff --git a/service/systemd/mosquitto.service.notify b/service/systemd/mosquitto.service.notify index 584a1101..84b08f84 100644 --- a/service/systemd/mosquitto.service.notify +++ b/service/systemd/mosquitto.service.notify @@ -1,5 +1,5 @@ [Unit] -Description=Mosquitto MQTT v3.1/v3.1.1 Broker +Description=Mosquitto MQTT Broker Documentation=man:mosquitto.conf(5) man:mosquitto(8) After=network.target Wants=network.target @@ -10,6 +10,8 @@ NotifyAccess=main ExecStart=/usr/sbin/mosquitto -c /etc/mosquitto/mosquitto.conf ExecReload=/bin/kill -HUP $MAINPID Restart=on-failure +ExecStartPre=/bin/mkdir -m 740 -p /var/log/mosquitto +ExecStartPre=/bin/chown mosquitto: /var/log/mosquitto [Install] WantedBy=multi-user.target diff --git a/service/systemd/mosquitto.service.simple b/service/systemd/mosquitto.service.simple index 4346ea14..d0e657e6 100644 --- a/service/systemd/mosquitto.service.simple +++ b/service/systemd/mosquitto.service.simple @@ -1,5 +1,5 @@ [Unit] -Description=Mosquitto MQTT v3.1/v3.1.1 Broker +Description=Mosquitto MQTT Broker Documentation=man:mosquitto.conf(5) man:mosquitto(8) After=network.target Wants=network.target @@ -8,6 +8,8 @@ Wants=network.target ExecStart=/usr/sbin/mosquitto -c /etc/mosquitto/mosquitto.conf ExecReload=/bin/kill -HUP $MAINPID Restart=on-failure +ExecStartPre=/bin/mkdir -m 740 -p /var/log/mosquitto +ExecStartPre=/bin/chown mosquitto: /var/log/mosquitto [Install] WantedBy=multi-user.target diff --git a/set-version.sh b/set-version.sh index 2d081eeb..c98f34bd 100755 --- a/set-version.sh +++ b/set-version.sh @@ -2,7 +2,7 @@ MAJOR=1 MINOR=6 -REVISION=10 +REVISION=11 sed -i "s/^VERSION=.*/VERSION=${MAJOR}.${MINOR}.${REVISION}/" config.mk diff --git a/snap/snapcraft.yaml b/snap/snapcraft.yaml index 4de5f916..a8ba134a 100644 --- a/snap/snapcraft.yaml +++ b/snap/snapcraft.yaml @@ -1,5 +1,5 @@ name: mosquitto -version: 1.6.10 +version: 1.6.11 summary: Eclipse Mosquitto MQTT broker description: This is a message broker that supports version 5.0, 3.1.1, and 3.1 of the MQTT protocol. diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index ee34673b..053ca1d9 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -150,6 +150,8 @@ endif (HAVE_GETADDRINFO_A AND WITH_ADNS) if (UNIX) if (APPLE) set (MOSQ_LIBS ${MOSQ_LIBS} dl m) + elseif (${CMAKE_SYSTEM_NAME} MATCHES "OpenBSD") + set (MOSQ_LIBS ${MOSQ_LIBS} m) elseif(QNX) set(MOSQ_LIBS ${MOSQ_LIBS} m socket) else(APPLE) diff --git a/src/conf.c b/src/conf.c index ea5651f8..6ec461ed 100644 --- a/src/conf.c +++ b/src/conf.c @@ -356,7 +356,7 @@ void config__cleanup(struct mosquitto__config *config) static void print_usage(void) { printf("mosquitto version %s\n\n", VERSION); - printf("mosquitto is an MQTT v3.1.1 broker.\n\n"); + printf("mosquitto is an MQTT v5.0/v3.1.1/v3.1 broker.\n\n"); printf("Usage: mosquitto [-c config_file] [-d] [-h] [-p port]\n\n"); printf(" -c : specify the broker config file.\n"); printf(" -d : put the broker into the background after starting.\n"); @@ -365,7 +365,7 @@ static void print_usage(void) printf(" Not recommended in conjunction with the -c option.\n"); printf(" -v : verbose mode - enable all logging types. This overrides\n"); printf(" any logging options given in the config file.\n"); - printf("\nSee http://mosquitto.org/ for more information.\n\n"); + printf("\nSee https://mosquitto.org/ for more information.\n\n"); } int config__parse_args(struct mosquitto_db *db, struct mosquitto__config *config, int argc, char *argv[]) @@ -653,7 +653,7 @@ int config__read(struct mosquitto_db *db, struct mosquitto__config *config, bool } /* Check plugins loaded to see if they have username/password checks enabled */ - for(j=0; jlisteners[i].security_options.auth_plugin_config_count; j++){ + for(j=0; jlisteners[i].security_options.auth_plugin_config_count; j++){ plugin = &config->listeners[i].security_options.auth_plugin_configs[j].plugin; if(plugin->version == 3 || plugin->version == 2){ @@ -685,7 +685,7 @@ int config__read(struct mosquitto_db *db, struct mosquitto__config *config, bool } /* Check plugins loaded to see if they have username/password checks enabled */ - for(j=0; jsecurity_options.auth_plugin_config_count; j++){ + for(j=0; jsecurity_options.auth_plugin_config_count; j++){ plugin = &config->security_options.auth_plugin_configs[j].plugin; if(plugin->version == 3 || plugin->version == 2){ @@ -759,6 +759,76 @@ int config__read(struct mosquitto_db *db, struct mosquitto__config *config, bool return MOSQ_ERR_SUCCESS; } + +static int config__create_bridge_remap_topic(const char *prefix, const char *topic, char **remap_topic) +{ + int len; + + if(prefix){ + if(topic){ + len = strlen(topic) + strlen(prefix)+1; + *remap_topic = mosquitto__malloc(len+1); + if(!(*remap_topic)){ + log__printf(NULL, MOSQ_LOG_ERR, "Error: Out of memory."); + return MOSQ_ERR_NOMEM; + } + snprintf(*remap_topic, len+1, "%s%s", prefix, topic); + (*remap_topic)[len] = '\0'; + }else{ + *remap_topic = mosquitto__strdup(prefix); + if(!(*remap_topic)){ + log__printf(NULL, MOSQ_LOG_ERR, "Error: Out of memory."); + return MOSQ_ERR_NOMEM; + } + } + }else{ + *remap_topic = mosquitto__strdup(topic); + if(!(*remap_topic)){ + log__printf(NULL, MOSQ_LOG_ERR, "Error: Out of memory."); + return MOSQ_ERR_NOMEM; + } + } + return MOSQ_ERR_SUCCESS; +} + + +static int config__create_bridge_prefix(char **prefix, const char *topic, const char *token, const char *direction) +{ + int len; + + if(topic){ + len = strlen(topic) + strlen(token) + 1; + }else{ + len = strlen(token) + 1; + } + *prefix = malloc(len); + if(*prefix == NULL){ + log__printf(NULL, MOSQ_LOG_ERR, "Error: Out of memory."); + return MOSQ_ERR_NOMEM; + } + + if(topic){ + /* Print prefix+pattern to check for validity */ + snprintf(*prefix, len, "%s%s", token, topic); + }else{ + snprintf(*prefix, len, "%s", token); + } + + if(mosquitto_sub_topic_check(*prefix) != MOSQ_ERR_SUCCESS){ + log__printf(NULL, MOSQ_LOG_ERR, + "Error: Invalid bridge topic %s prefix and pattern combination '%s'.", + direction, *prefix); + + return MOSQ_ERR_INVAL; + } + + /* Print just the prefix for storage */ + snprintf(*prefix, len, "%s", token); + + return MOSQ_ERR_SUCCESS; +} + + int config__read_file_core(struct mosquitto__config *config, bool reload, struct config_recurse *cr, int level, int *lineno, FILE *fptr, char **buf, int *buflen) { int rc; @@ -769,7 +839,6 @@ int config__read_file_core(struct mosquitto__config *config, bool reload, struct char *tmp_char; struct mosquitto__bridge *cur_bridge = NULL; struct mosquitto__bridge_topic *cur_topic; - int len; #endif struct mosquitto__auth_plugin_config *cur_auth_plugin_config = NULL; @@ -2021,22 +2090,12 @@ int config__read_file_core(struct mosquitto__config *config, bool reload, struct log__printf(NULL, MOSQ_LOG_ERR, "Error: Invalid bridge topic local prefix '%s'.", token); return MOSQ_ERR_INVAL; } - cur_topic->local_prefix = malloc(strlen(cur_topic->topic) + strlen(token) + 1); - if(cur_topic->local_prefix == NULL){ - log__printf(NULL, MOSQ_LOG_ERR, "Error: Out of memory."); - return MOSQ_ERR_NOMEM; - } - /* Print prefix+pattern to check for validity */ - snprintf(cur_topic->local_prefix, strlen(cur_topic->topic) + strlen(token)+1, - "%s%s", token, cur_topic->topic); - if(mosquitto_sub_topic_check(cur_topic->local_prefix) != MOSQ_ERR_SUCCESS){ - log__printf(NULL, MOSQ_LOG_ERR, "Error: Invalid bridge topic local prefix and pattern combination '%s'.", cur_topic->local_prefix); + + if(config__create_bridge_prefix(&cur_topic->local_prefix, + cur_topic->topic, token, "local")){ + return MOSQ_ERR_INVAL; } - - /* Print just the prefix for storage */ - snprintf(cur_topic->local_prefix, strlen(cur_topic->topic) + strlen(token)+1, - "%s", token); } token = strtok_r(NULL, " ", &saveptr); @@ -2048,24 +2107,11 @@ int config__read_file_core(struct mosquitto__config *config, bool reload, struct log__printf(NULL, MOSQ_LOG_ERR, "Error: Invalid bridge topic remote prefix '%s'.", token); return MOSQ_ERR_INVAL; } - cur_topic->remote_prefix = malloc(strlen(cur_topic->topic) + strlen(token) + 1); - if(cur_topic == NULL){ - log__printf(NULL, MOSQ_LOG_ERR, "Error: Out of memory."); - return MOSQ_ERR_NOMEM; - } - /* Print prefix+pattern to check for validity */ - snprintf(cur_topic->remote_prefix, strlen(cur_topic->topic) + strlen(token)+1, - "%s%s", token, cur_topic->topic); - if(mosquitto_sub_topic_check(cur_topic->remote_prefix) != MOSQ_ERR_SUCCESS){ - log__printf(NULL, MOSQ_LOG_ERR, - "Error: Invalid bridge topic remote prefix and pattern combination '%s'.", - cur_topic->remote_prefix); + if(config__create_bridge_prefix(&cur_topic->remote_prefix, + cur_topic->topic, token, "remote")){ + return MOSQ_ERR_INVAL; } - - /* Print just the prefix for storage */ - snprintf(cur_topic->remote_prefix, strlen(cur_topic->topic) + strlen(token)+1, - "%s", token); } } } @@ -2077,54 +2123,16 @@ int config__read_file_core(struct mosquitto__config *config, bool reload, struct log__printf(NULL, MOSQ_LOG_ERR, "Error: Invalid bridge remapping."); return MOSQ_ERR_INVAL; } - if(cur_topic->local_prefix){ - if(cur_topic->topic){ - len = strlen(cur_topic->topic) + strlen(cur_topic->local_prefix)+1; - cur_topic->local_topic = mosquitto__malloc(len+1); - if(!cur_topic->local_topic){ - log__printf(NULL, MOSQ_LOG_ERR, "Error: Out of memory."); - return MOSQ_ERR_NOMEM; - } - snprintf(cur_topic->local_topic, len+1, "%s%s", cur_topic->local_prefix, cur_topic->topic); - cur_topic->local_topic[len] = '\0'; - }else{ - cur_topic->local_topic = mosquitto__strdup(cur_topic->local_prefix); - if(!cur_topic->local_topic){ - log__printf(NULL, MOSQ_LOG_ERR, "Error: Out of memory."); - return MOSQ_ERR_NOMEM; - } - } - }else{ - cur_topic->local_topic = mosquitto__strdup(cur_topic->topic); - if(!cur_topic->local_topic){ - log__printf(NULL, MOSQ_LOG_ERR, "Error: Out of memory."); - return MOSQ_ERR_NOMEM; - } + if(config__create_bridge_remap_topic(cur_topic->local_prefix, + cur_topic->topic, &cur_topic->local_topic)){ + + return MOSQ_ERR_INVAL; } - if(cur_topic->remote_prefix){ - if(cur_topic->topic){ - len = strlen(cur_topic->topic) + strlen(cur_topic->remote_prefix)+1; - cur_topic->remote_topic = mosquitto__malloc(len+1); - if(!cur_topic->remote_topic){ - log__printf(NULL, MOSQ_LOG_ERR, "Error: Out of memory."); - return MOSQ_ERR_NOMEM; - } - snprintf(cur_topic->remote_topic, len, "%s%s", cur_topic->remote_prefix, cur_topic->topic); - cur_topic->remote_topic[len] = '\0'; - }else{ - cur_topic->remote_topic = mosquitto__strdup(cur_topic->remote_prefix); - if(!cur_topic->remote_topic){ - log__printf(NULL, MOSQ_LOG_ERR, "Error: Out of memory."); - return MOSQ_ERR_NOMEM; - } - } - }else{ - cur_topic->remote_topic = mosquitto__strdup(cur_topic->topic); - if(!cur_topic->remote_topic){ - log__printf(NULL, MOSQ_LOG_ERR, "Error: Out of memory."); - return MOSQ_ERR_NOMEM; - } + if(config__create_bridge_remap_topic(cur_topic->remote_prefix, + cur_topic->topic, &cur_topic->remote_topic)){ + + return MOSQ_ERR_INVAL; } #else log__printf(NULL, MOSQ_LOG_WARNING, "Warning: Bridge support not available."); diff --git a/src/database.c b/src/database.c index 11980d7a..124824cd 100644 --- a/src/database.c +++ b/src/database.c @@ -339,6 +339,9 @@ int db__message_delete_outgoing(struct mosquitto_db *db, struct mosquitto *conte } db__message_dequeue_first(context, &context->msgs_out); } +#ifdef WITH_PERSISTENCE + db->persistence_changes++; +#endif return MOSQ_ERR_SUCCESS; } @@ -394,6 +397,10 @@ int db__message_insert(struct mosquitto_db *db, struct mosquitto *context, uint1 } } } + if(context->bridge && context->bridge->clean_start == true){ + mosquitto_property_free_all(&properties); + return 2; + } } if(context->sock != INVALID_SOCKET){ diff --git a/src/db_dump/db_dump.c b/src/db_dump/db_dump.c index 1488f2ef..c453ed99 100644 --- a/src/db_dump/db_dump.c +++ b/src/db_dump/db_dump.c @@ -110,8 +110,8 @@ static int dump__cfg_chunk_process(struct mosquitto_db *db, FILE *db_fd, uint32_ memset(&chunk, 0, sizeof(struct PF_cfg)); - if(db_version == 5){ - rc = persist__chunk_cfg_read_v5(db_fd, &chunk); + if(db_version == 6 || db_version == 5){ + rc = persist__chunk_cfg_read_v56(db_fd, &chunk); }else{ rc = persist__chunk_cfg_read_v234(db_fd, &chunk); } @@ -147,8 +147,8 @@ static int dump__client_chunk_process(struct mosquitto_db *db, FILE *db_fd, uint memset(&chunk, 0, sizeof(struct P_client)); - if(db_version == 5){ - rc = persist__chunk_client_read_v5(db_fd, &chunk); + if(db_version == 6 || db_version == 5){ + rc = persist__chunk_client_read_v56(db_fd, &chunk, db_version); }else{ rc = persist__chunk_client_read_v234(db_fd, &chunk, db_version); } @@ -189,8 +189,8 @@ static int dump__client_msg_chunk_process(struct mosquitto_db *db, FILE *db_fd, client_msg_count++; memset(&chunk, 0, sizeof(struct P_client_msg)); - if(db_version == 5){ - rc = persist__chunk_client_msg_read_v5(db_fd, &chunk, length); + if(db_version == 6 || db_version == 5){ + rc = persist__chunk_client_msg_read_v56(db_fd, &chunk, length); }else{ rc = persist__chunk_client_msg_read_v234(db_fd, &chunk); } @@ -234,8 +234,8 @@ static int dump__msg_store_chunk_process(struct mosquitto_db *db, FILE *db_fptr, msg_store_count++; memset(&chunk, 0, sizeof(struct P_msg_store)); - if(db_version == 5){ - rc = persist__chunk_msg_store_read_v5(db_fptr, &chunk, length); + if(db_version == 6 || db_version == 5){ + rc = persist__chunk_msg_store_read_v56(db_fptr, &chunk, length); }else{ rc = persist__chunk_msg_store_read_v234(db_fptr, &chunk, db_version); } @@ -321,8 +321,8 @@ static int dump__retain_chunk_process(struct mosquitto_db *db, FILE *db_fd, uint if(do_print) printf("DB_CHUNK_RETAIN:\n"); if(do_print) printf("\tLength: %d\n", length); - if(db_version == 5){ - rc = persist__chunk_retain_read_v5(db_fd, &chunk); + if(db_version == 6 || db_version == 5){ + rc = persist__chunk_retain_read_v56(db_fd, &chunk); }else{ rc = persist__chunk_retain_read_v234(db_fd, &chunk); } @@ -345,8 +345,8 @@ static int dump__sub_chunk_process(struct mosquitto_db *db, FILE *db_fd, uint32_ sub_count++; memset(&chunk, 0, sizeof(struct P_sub)); - if(db_version == 5){ - rc = persist__chunk_sub_read_v5(db_fd, &chunk); + if(db_version == 6 || db_version == 5){ + rc = persist__chunk_sub_read_v56(db_fd, &chunk); }else{ rc = persist__chunk_sub_read_v234(db_fd, &chunk); } diff --git a/src/db_dump/print.c b/src/db_dump/print.c index 8effd317..414f931d 100644 --- a/src/db_dump/print.c +++ b/src/db_dump/print.c @@ -145,6 +145,12 @@ void print__client(struct P_client *chunk, int length) printf("DB_CHUNK_CLIENT:\n"); printf("\tLength: %d\n", length); printf("\tClient ID: %s\n", chunk->client_id); + if(chunk->username){ + printf("\tUsername: %s\n", chunk->username); + } + if(chunk->F.listener_port > 0){ + printf("\tListener port: %u\n", chunk->F.listener_port); + } printf("\tLast MID: %d\n", chunk->F.last_mid); printf("\tSession expiry time: %" PRIu64 "\n", chunk->F.session_expiry_time); printf("\tSession expiry interval: %u\n", chunk->F.session_expiry_interval); diff --git a/src/handle_publish.c b/src/handle_publish.c index 2504860d..acbbec93 100644 --- a/src/handle_publish.c +++ b/src/handle_publish.c @@ -109,7 +109,10 @@ int handle__publish(struct mosquitto_db *db, struct mosquitto *context) /* Handle properties */ if(context->protocol == mosq_p_mqtt5){ rc = property__read_all(CMD_PUBLISH, &context->in_packet, &properties); - if(rc) return rc; + if(rc){ + mosquitto__free(topic); + return rc; + } p = properties; p_prev = NULL; diff --git a/src/handle_subscribe.c b/src/handle_subscribe.c index 2ebf8ae4..c42604ba 100644 --- a/src/handle_subscribe.c +++ b/src/handle_subscribe.c @@ -119,6 +119,8 @@ int handle__subscribe(struct mosquitto_db *db, struct mosquitto *context) retain_handling = (subscription_options & 0x30); if(retain_handling == 0x30 || (subscription_options & 0xC0) != 0){ + mosquitto__free(sub); + mosquitto__free(payload); return MOSQ_ERR_PROTOCOL; } } diff --git a/src/handle_unsubscribe.c b/src/handle_unsubscribe.c index 9869b32a..df27911d 100644 --- a/src/handle_unsubscribe.c +++ b/src/handle_unsubscribe.c @@ -49,7 +49,7 @@ int handle__unsubscribe(struct mosquitto_db *db, struct mosquitto *context) return MOSQ_ERR_PROTOCOL; } } - if(packet__read_uint16(&context->in_packet, &mid)) return 1; + if(packet__read_uint16(&context->in_packet, &mid)) return MOSQ_ERR_PROTOCOL; if(mid == 0) return MOSQ_ERR_PROTOCOL; if(context->protocol == mosq_p_mqtt5){ @@ -76,7 +76,7 @@ int handle__unsubscribe(struct mosquitto_db *db, struct mosquitto *context) sub = NULL; if(packet__read_string(&context->in_packet, &sub, &slen)){ mosquitto__free(reason_codes); - return 1; + return MOSQ_ERR_PROTOCOL; } if(!slen){ @@ -85,7 +85,7 @@ int handle__unsubscribe(struct mosquitto_db *db, struct mosquitto *context) context->id); mosquitto__free(sub); mosquitto__free(reason_codes); - return 1; + return MOSQ_ERR_PROTOCOL; } if(mosquitto_sub_topic_check(sub)){ log__printf(NULL, MOSQ_LOG_INFO, @@ -93,7 +93,7 @@ int handle__unsubscribe(struct mosquitto_db *db, struct mosquitto *context) context->id); mosquitto__free(sub); mosquitto__free(reason_codes); - return 1; + return MOSQ_ERR_PROTOCOL; } log__printf(NULL, MOSQ_LOG_DEBUG, "\t%s", sub); diff --git a/src/logging.c b/src/logging.c index a43ca22e..ac079a85 100644 --- a/src/logging.c +++ b/src/logging.c @@ -24,6 +24,7 @@ Contributors: #include #ifdef WITH_DLT +#include #include #endif @@ -55,6 +56,29 @@ static int log_priorities = MOSQ_LOG_ERR | MOSQ_LOG_WARNING | MOSQ_LOG_NOTICE | #ifdef WITH_DLT static DltContext dltContext; +static bool dlt_allowed = false; + +void dlt_fifo_check(void) +{ + struct stat statbuf; + int fd; + + /* If we start DLT but the /tmp/dlt fifo doesn't exist, or isn't available + * for writing then there is a big delay when we try and close the log + * later, so check for it first. This has the side effect of not letting + * people using DLT create the fifo after Mosquitto has started, but at the + * benefit of not having a massive delay for everybody else. */ + memset(&statbuf, 0, sizeof(statbuf)); + if(stat("/tmp/dlt", &statbuf) == 0){ + if(S_ISFIFO(statbuf.st_mode)){ + fd = open("/tmp/dlt", O_NONBLOCK | O_WRONLY); + if(fd != -1){ + dlt_allowed = true; + close(fd); + } + } + } +} #endif static int get_time(struct tm **ti) @@ -110,17 +134,21 @@ int log__init(struct mosquitto__config *config) return 1; } config->log_fptr = mosquitto__fopen(config->log_file, "at", true); - if(!config->log_fptr){ + if(config->log_fptr){ + setvbuf(config->log_fptr, NULL, _IOLBF, 0); + }else{ log_destinations = MQTT3_LOG_STDERR; log_priorities = MOSQ_LOG_ERR; log__printf(NULL, MOSQ_LOG_ERR, "Error: Unable to open log file %s for writing.", config->log_file); - return MOSQ_ERR_INVAL; } restore_privileges(); } #ifdef WITH_DLT - DLT_REGISTER_APP("MQTT","mosquitto log"); - dlt_register_context(&dltContext, "MQTT", "mosquitto DLT context"); + dlt_fifo_check(); + if(dlt_allowed){ + DLT_REGISTER_APP("MQTT","mosquitto log"); + dlt_register_context(&dltContext, "MQTT", "mosquitto DLT context"); + } #endif return rc; } @@ -142,8 +170,10 @@ int log__close(struct mosquitto__config *config) } #ifdef WITH_DLT - dlt_unregister_context(&dltContext); - DLT_UNREGISTER_APP(); + if(dlt_allowed){ + dlt_unregister_context(&dltContext); + DLT_UNREGISTER_APP(); + } #endif /* FIXME - do something for all destinations! */ return MOSQ_ERR_SUCCESS; @@ -182,7 +212,6 @@ int log__vprintf(int priority, const char *fmt, va_list va) const char *topic; int syslog_priority; time_t now = time(NULL); - static time_t last_flush = 0; char time_buf[50]; bool log_timestamp = true; char *log_timestamp_format = NULL; @@ -294,7 +323,6 @@ int log__vprintf(int priority, const char *fmt, va_list va) }else{ fprintf(stdout, "%s\n", s); } - fflush(stdout); } if(log_destinations & MQTT3_LOG_STDERR){ if(log_timestamp){ @@ -306,7 +334,6 @@ int log__vprintf(int priority, const char *fmt, va_list va) }else{ fprintf(stderr, "%s\n", s); } - fflush(stderr); } if(log_destinations & MQTT3_LOG_FILE && log_fptr){ if(log_timestamp){ @@ -318,10 +345,6 @@ int log__vprintf(int priority, const char *fmt, va_list va) }else{ fprintf(log_fptr, "%s\n", s); } - if(now - last_flush > 1){ - fflush(log_fptr); - last_flush = now; - } } if(log_destinations & MQTT3_LOG_SYSLOG){ #ifndef WIN32 diff --git a/src/mosquitto.c b/src/mosquitto.c index 43b0a203..985c31d6 100644 --- a/src/mosquitto.c +++ b/src/mosquitto.c @@ -289,6 +289,18 @@ int main(int argc, char *argv[]) rc = mosquitto_security_init(&int_db, false); if(rc) return rc; + /* After loading persisted clients and ACLs, try to associate them, + * so persisted subscriptions can start storing messages */ + HASH_ITER(hh_id, int_db.contexts_by_id, ctxt, ctxt_tmp){ + if(ctxt && !ctxt->clean_start && ctxt->username){ + rc = acl__find_acls(&int_db, ctxt); + if(rc){ + log__printf(NULL, MOSQ_LOG_WARNING, "Failed to associate persisted user %s with ACLs, " + "likely due to changed ports while using a per_listener_settings configuration.", ctxt->username); + } + } + } + #ifdef WITH_SYS_TREE sys_tree__init(&int_db); #endif @@ -333,10 +345,6 @@ int main(int argc, char *argv[]) #endif } } - if(listensock == NULL){ - log__printf(NULL, MOSQ_LOG_ERR, "Error: Unable to start any listening sockets, exiting."); - return 1; - } rc = drop_privileges(&config, false); if(rc != MOSQ_ERR_SUCCESS) return rc; @@ -364,6 +372,7 @@ int main(int argc, char *argv[]) } #endif + log__printf(NULL, MOSQ_LOG_INFO, "mosquitto version %s running", VERSION); #ifdef WITH_SYSTEMD sd_notify(0, "READY=1"); #endif diff --git a/src/net.c b/src/net.c index 169c0a99..3833b947 100644 --- a/src/net.c +++ b/src/net.c @@ -86,7 +86,7 @@ static void net__print_error(int log, const char *format_str) #ifdef WIN32 FormatMessage(FORMAT_MESSAGE_ALLOCATE_BUFFER | FORMAT_MESSAGE_FROM_SYSTEM, - NULL, WSAGetLastError(), LANG_NEUTRAL, &buf, 0, NULL); + NULL, WSAGetLastError(), LANG_NEUTRAL, (LPTSTR)&buf, 0, NULL); log__printf(NULL, log, format_str, buf); LocalFree(buf); @@ -165,7 +165,11 @@ int net__socket_accept(struct mosquitto_db *db, mosq_sock_t listensock) if(db->config->set_tcp_nodelay){ int flag = 1; +#ifdef WIN32 + if (setsockopt(new_sock, IPPROTO_TCP, TCP_NODELAY, (char *)&flag, sizeof(int)) != 0) { +#else if(setsockopt(new_sock, IPPROTO_TCP, TCP_NODELAY, &flag, sizeof(int)) != 0){ +#endif log__printf(NULL, MOSQ_LOG_WARNING, "Warning: Unable to set TCP_NODELAY."); } } @@ -588,11 +592,7 @@ int net__socket_listen(struct mosquitto__listener *listener) struct addrinfo *ainfo, *rp; char service[10]; int rc; -#ifndef WIN32 int ss_opt = 1; -#else - char ss_opt = 1; -#endif #ifdef SO_BINDTODEVICE struct ifreq ifr; #endif diff --git a/src/persist.h b/src/persist.h index b9bc3708..816ec400 100644 --- a/src/persist.h +++ b/src/persist.h @@ -17,7 +17,7 @@ Contributors: #ifndef PERSIST_H #define PERSIST_H -#define MOSQ_DB_VERSION 5 +#define MOSQ_DB_VERSION 6 /* DB read/write */ extern const unsigned char magic[15]; @@ -57,15 +57,28 @@ struct PF_cfg{ uint8_t dbid_size; }; -struct PF_client{ +struct PF_client_v5{ int64_t session_expiry_time; uint32_t session_expiry_interval; uint16_t last_mid; uint16_t id_len; }; +struct PF_client{ + /* struct PF_client_v5; */ + int64_t session_expiry_time; + uint32_t session_expiry_interval; + uint16_t last_mid; + uint16_t id_len; + + uint16_t listener_port; + uint16_t username_len; + /* tail: 4 byte padding, because 64bit member + * forces multiple of 8 for struct size */ +}; struct P_client{ struct PF_client F; char *client_id; + char *username; }; @@ -141,19 +154,19 @@ int persist__chunk_msg_store_read_v234(FILE *db_fptr, struct P_msg_store *chunk, int persist__chunk_retain_read_v234(FILE *db_fptr, struct P_retain *chunk); int persist__chunk_sub_read_v234(FILE *db_fptr, struct P_sub *chunk); -int persist__chunk_header_read_v5(FILE *db_fptr, int *chunk, int *length); -int persist__chunk_cfg_read_v5(FILE *db_fptr, struct PF_cfg *chunk); -int persist__chunk_client_read_v5(FILE *db_fptr, struct P_client *chunk); -int persist__chunk_client_msg_read_v5(FILE *db_fptr, struct P_client_msg *chunk, uint32_t length); -int persist__chunk_msg_store_read_v5(FILE *db_fptr, struct P_msg_store *chunk, uint32_t length); -int persist__chunk_retain_read_v5(FILE *db_fptr, struct P_retain *chunk); -int persist__chunk_sub_read_v5(FILE *db_fptr, struct P_sub *chunk); +int persist__chunk_header_read_v56(FILE *db_fptr, int *chunk, int *length); +int persist__chunk_cfg_read_v56(FILE *db_fptr, struct PF_cfg *chunk); +int persist__chunk_client_read_v56(FILE *db_fptr, struct P_client *chunk, int db_version); +int persist__chunk_client_msg_read_v56(FILE *db_fptr, struct P_client_msg *chunk, uint32_t length); +int persist__chunk_msg_store_read_v56(FILE *db_fptr, struct P_msg_store *chunk, uint32_t length); +int persist__chunk_retain_read_v56(FILE *db_fptr, struct P_retain *chunk); +int persist__chunk_sub_read_v56(FILE *db_fptr, struct P_sub *chunk); -int persist__chunk_cfg_write_v5(FILE *db_fptr, struct PF_cfg *chunk); -int persist__chunk_client_write_v5(FILE *db_fptr, struct P_client *chunk); -int persist__chunk_client_msg_write_v5(FILE *db_fptr, struct P_client_msg *chunk); -int persist__chunk_message_store_write_v5(FILE *db_fptr, struct P_msg_store *chunk); -int persist__chunk_retain_write_v5(FILE *db_fptr, struct P_retain *chunk); -int persist__chunk_sub_write_v5(FILE *db_fptr, struct P_sub *chunk); +int persist__chunk_cfg_write_v6(FILE *db_fptr, struct PF_cfg *chunk); +int persist__chunk_client_write_v6(FILE *db_fptr, struct P_client *chunk); +int persist__chunk_client_msg_write_v6(FILE *db_fptr, struct P_client_msg *chunk); +int persist__chunk_message_store_write_v6(FILE *db_fptr, struct P_msg_store *chunk); +int persist__chunk_retain_write_v6(FILE *db_fptr, struct P_retain *chunk); +int persist__chunk_sub_write_v6(FILE *db_fptr, struct P_sub *chunk); #endif diff --git a/src/persist_read.c b/src/persist_read.c index ad0c78e3..df643020 100644 --- a/src/persist_read.c +++ b/src/persist_read.c @@ -175,14 +175,14 @@ static int persist__client_msg_restore(struct mosquitto_db *db, struct P_client_ static int persist__client_chunk_restore(struct mosquitto_db *db, FILE *db_fptr) { - int rc = 0; + int i, rc = 0; struct mosquitto *context; struct P_client chunk; memset(&chunk, 0, sizeof(struct P_client)); - if(db_version == 5){ - rc = persist__chunk_client_read_v5(db_fptr, &chunk); + if(db_version == 6 || db_version == 5){ + rc = persist__chunk_client_read_v56(db_fptr, &chunk, db_version); }else{ rc = persist__chunk_client_read_v234(db_fptr, &chunk, db_version); } @@ -195,13 +195,29 @@ static int persist__client_chunk_restore(struct mosquitto_db *db, FILE *db_fptr) if(context){ context->session_expiry_time = chunk.F.session_expiry_time; context->session_expiry_interval = chunk.F.session_expiry_interval; + if(chunk.username && !context->username){ + /* username is not freed here, it is now owned by context */ + context->username = chunk.username; + chunk.username = NULL; + /* in per_listener_settings mode, try to find the listener by persisted port */ + if(db->config->per_listener_settings && !context->listener && chunk.F.listener_port > 0){ + for(i=0; i < db->config->listener_count; i++){ + if(db->config->listeners[i].port == chunk.F.listener_port){ + context->listener = &db->config->listeners[i]; + break; + } + } + } + } /* FIXME - we should expire clients here if they have exceeded their time */ }else{ rc = 1; } mosquitto__free(chunk.client_id); - + if(chunk.username){ + mosquitto__free(chunk.username); + } return rc; } @@ -213,8 +229,8 @@ static int persist__client_msg_chunk_restore(struct mosquitto_db *db, FILE *db_f memset(&chunk, 0, sizeof(struct P_client_msg)); - if(db_version == 5){ - rc = persist__chunk_client_msg_read_v5(db_fptr, &chunk, length); + if(db_version == 6 || db_version == 5){ + rc = persist__chunk_client_msg_read_v56(db_fptr, &chunk, length); }else{ rc = persist__chunk_client_msg_read_v234(db_fptr, &chunk); } @@ -242,8 +258,8 @@ static int persist__msg_store_chunk_restore(struct mosquitto_db *db, FILE *db_fp memset(&chunk, 0, sizeof(struct P_msg_store)); - if(db_version == 5){ - rc = persist__chunk_msg_store_read_v5(db_fptr, &chunk, length); + if(db_version == 6 || db_version == 5){ + rc = persist__chunk_msg_store_read_v56(db_fptr, &chunk, length); }else{ rc = persist__chunk_msg_store_read_v234(db_fptr, &chunk, db_version); } @@ -320,8 +336,8 @@ static int persist__retain_chunk_restore(struct mosquitto_db *db, FILE *db_fptr) memset(&chunk, 0, sizeof(struct P_retain)); - if(db_version == 5){ - rc = persist__chunk_retain_read_v5(db_fptr, &chunk); + if(db_version == 6 || db_version == 5){ + rc = persist__chunk_retain_read_v56(db_fptr, &chunk); }else{ rc = persist__chunk_retain_read_v234(db_fptr, &chunk); } @@ -346,8 +362,8 @@ static int persist__sub_chunk_restore(struct mosquitto_db *db, FILE *db_fptr) memset(&chunk, 0, sizeof(struct P_sub)); - if(db_version == 5){ - rc = persist__chunk_sub_read_v5(db_fptr, &chunk); + if(db_version == 6 || db_version == 5){ + rc = persist__chunk_sub_read_v56(db_fptr, &chunk); }else{ rc = persist__chunk_sub_read_v234(db_fptr, &chunk); } @@ -367,8 +383,8 @@ static int persist__sub_chunk_restore(struct mosquitto_db *db, FILE *db_fptr) int persist__chunk_header_read(FILE *db_fptr, int *chunk, int *length) { - if(db_version == 5){ - return persist__chunk_header_read_v5(db_fptr, chunk, length); + if(db_version == 6 || db_version == 5){ + return persist__chunk_header_read_v56(db_fptr, chunk, length); }else{ return persist__chunk_header_read_v234(db_fptr, chunk, length); } @@ -416,7 +432,9 @@ int persist__restore(struct mosquitto_db *db) * Is your DB change still compatible with previous versions? */ if(db_version != MOSQ_DB_VERSION){ - if(db_version == 4){ + if(db_version == 5){ + /* Addition of username and listener_port to client chunk in v6 */ + }else if(db_version == 4){ }else if(db_version == 3){ /* Addition of source_username and source_port to msg_store chunk in v4, v1.5.6 */ }else if(db_version == 2){ @@ -431,8 +449,8 @@ int persist__restore(struct mosquitto_db *db) while(persist__chunk_header_read(fptr, &chunk, &length) == MOSQ_ERR_SUCCESS){ switch(chunk){ case DB_CHUNK_CFG: - if(db_version == 5){ - if(persist__chunk_cfg_read_v5(fptr, &cfg_chunk)){ + if(db_version == 6 || db_version == 5){ + if(persist__chunk_cfg_read_v56(fptr, &cfg_chunk)){ fclose(fptr); return 1; } diff --git a/src/persist_read_v5.c b/src/persist_read_v5.c index e8bb5aed..0d0605b1 100644 --- a/src/persist_read_v5.c +++ b/src/persist_read_v5.c @@ -38,7 +38,7 @@ Contributors: #include "util_mosq.h" -int persist__chunk_header_read_v5(FILE *db_fptr, int *chunk, int *length) +int persist__chunk_header_read_v56(FILE *db_fptr, int *chunk, int *length) { size_t rlen; struct PF_header header; @@ -53,7 +53,7 @@ int persist__chunk_header_read_v5(FILE *db_fptr, int *chunk, int *length) } -int persist__chunk_cfg_read_v5(FILE *db_fptr, struct PF_cfg *chunk) +int persist__chunk_cfg_read_v56(FILE *db_fptr, struct PF_cfg *chunk) { if(fread(chunk, sizeof(struct PF_cfg), 1, db_fptr) != 1){ log__printf(NULL, MOSQ_LOG_ERR, "Error: %s.", strerror(errno)); @@ -64,28 +64,46 @@ int persist__chunk_cfg_read_v5(FILE *db_fptr, struct PF_cfg *chunk) } -int persist__chunk_client_read_v5(FILE *db_fptr, struct P_client *chunk) +int persist__chunk_client_read_v56(FILE *db_fptr, struct P_client *chunk, int db_version) { int rc; - read_e(db_fptr, &chunk->F, sizeof(struct PF_client)); + if(db_version == 6){ + read_e(db_fptr, &chunk->F, sizeof(struct PF_client)); + chunk->F.username_len = ntohs(chunk->F.username_len); + chunk->F.listener_port = ntohs(chunk->F.listener_port); + }else if(db_version == 5){ + read_e(db_fptr, &chunk->F, sizeof(struct PF_client_v5)); + }else{ + return 1; + } + chunk->F.session_expiry_interval = ntohl(chunk->F.session_expiry_interval); chunk->F.last_mid = ntohs(chunk->F.last_mid); chunk->F.id_len = ntohs(chunk->F.id_len); + rc = persist__read_string_len(db_fptr, &chunk->client_id, chunk->F.id_len); if(rc || !chunk->client_id){ return 1; - }else{ - return MOSQ_ERR_SUCCESS; } + + if(chunk->F.username_len > 0){ + rc = persist__read_string_len(db_fptr, &chunk->username, chunk->F.username_len); + if(rc || !chunk->username){ + mosquitto__free(chunk->client_id); + return 1; + } + } + + return MOSQ_ERR_SUCCESS; error: log__printf(NULL, MOSQ_LOG_ERR, "Error: %s.", strerror(errno)); return 1; } -int persist__chunk_client_msg_read_v5(FILE *db_fptr, struct P_client_msg *chunk, uint32_t length) +int persist__chunk_client_msg_read_v56(FILE *db_fptr, struct P_client_msg *chunk, uint32_t length) { mosquitto_property *properties = NULL; struct mosquitto__packet prop_packet; @@ -125,7 +143,7 @@ error: } -int persist__chunk_msg_store_read_v5(FILE *db_fptr, struct P_msg_store *chunk, uint32_t length) +int persist__chunk_msg_store_read_v56(FILE *db_fptr, struct P_msg_store *chunk, uint32_t length) { int rc = 0; mosquitto_property *properties = NULL; @@ -215,7 +233,7 @@ error: } -int persist__chunk_retain_read_v5(FILE *db_fptr, struct P_retain *chunk) +int persist__chunk_retain_read_v56(FILE *db_fptr, struct P_retain *chunk) { if(fread(&chunk->F, sizeof(struct P_retain), 1, db_fptr) != 1){ log__printf(NULL, MOSQ_LOG_ERR, "Error: %s.", strerror(errno)); @@ -225,7 +243,7 @@ int persist__chunk_retain_read_v5(FILE *db_fptr, struct P_retain *chunk) } -int persist__chunk_sub_read_v5(FILE *db_fptr, struct P_sub *chunk) +int persist__chunk_sub_read_v56(FILE *db_fptr, struct P_sub *chunk) { int rc; diff --git a/src/persist_write.c b/src/persist_write.c index 61f4a73d..e6d25121 100644 --- a/src/persist_write.c +++ b/src/persist_write.c @@ -70,7 +70,7 @@ static int persist__client_messages_save(struct mosquitto_db *db, FILE *db_fptr, chunk.client_id = context->id; chunk.properties = cmsg->properties; - rc = persist__chunk_client_msg_write_v5(db_fptr, &chunk); + rc = persist__chunk_client_msg_write_v6(db_fptr, &chunk); if(rc){ return rc; } @@ -146,7 +146,7 @@ static int persist__message_store_save(struct mosquitto_db *db, FILE *db_fptr) chunk.payload = stored->payload; chunk.properties = stored->properties; - rc = persist__chunk_message_store_write_v5(db_fptr, &chunk); + rc = persist__chunk_message_store_write_v6(db_fptr, &chunk); if(rc){ return rc; } @@ -174,8 +174,15 @@ static int persist__client_save(struct mosquitto_db *db, FILE *db_fptr) chunk.F.last_mid = context->last_mid; chunk.F.id_len = strlen(context->id); chunk.client_id = context->id; + if(context->username){ + chunk.F.username_len = strlen(context->username); + chunk.username = context->username; + if(context->listener){ + chunk.F.listener_port = context->listener->port; + } + } - rc = persist__chunk_client_write_v5(db_fptr, &chunk); + rc = persist__chunk_client_write_v6(db_fptr, &chunk); if(rc){ return rc; } @@ -224,7 +231,7 @@ static int persist__subs_retain_save(struct mosquitto_db *db, FILE *db_fptr, str sub_chunk.client_id = sub->context->id; sub_chunk.topic = thistopic; - rc = persist__chunk_sub_write_v5(db_fptr, &sub_chunk); + rc = persist__chunk_sub_write_v6(db_fptr, &sub_chunk); if(rc){ mosquitto__free(thistopic); return rc; @@ -236,7 +243,7 @@ static int persist__subs_retain_save(struct mosquitto_db *db, FILE *db_fptr, str if(strncmp(node->retained->topic, "$SYS", 4)){ /* Don't save $SYS messages. */ retain_chunk.F.store_id = node->retained->db_id; - rc = persist__chunk_retain_write_v5(db_fptr, &retain_chunk); + rc = persist__chunk_retain_write_v6(db_fptr, &retain_chunk); if(rc){ mosquitto__free(thistopic); return rc; @@ -334,7 +341,7 @@ int persist__backup(struct mosquitto_db *db, bool shutdown) cfg_chunk.last_db_id = db->last_db_id; cfg_chunk.shutdown = shutdown; cfg_chunk.dbid_size = sizeof(dbid_t); - if(persist__chunk_cfg_write_v5(db_fptr, &cfg_chunk)){ + if(persist__chunk_cfg_write_v6(db_fptr, &cfg_chunk)){ goto error; } diff --git a/src/persist_write_v5.c b/src/persist_write_v5.c index ba960f09..48ca9c93 100644 --- a/src/persist_write_v5.c +++ b/src/persist_write_v5.c @@ -37,7 +37,7 @@ Contributors: #include "time_mosq.h" #include "util_mosq.h" -int persist__chunk_cfg_write_v5(FILE *db_fptr, struct PF_cfg *chunk) +int persist__chunk_cfg_write_v6(FILE *db_fptr, struct PF_cfg *chunk) { struct PF_header header; @@ -53,22 +53,28 @@ error: } -int persist__chunk_client_write_v5(FILE *db_fptr, struct P_client *chunk) +int persist__chunk_client_write_v6(FILE *db_fptr, struct P_client *chunk) { struct PF_header header; uint16_t id_len = chunk->F.id_len; + uint16_t username_len = chunk->F.username_len; chunk->F.session_expiry_interval = htonl(chunk->F.session_expiry_interval); chunk->F.last_mid = htons(chunk->F.last_mid); chunk->F.id_len = htons(chunk->F.id_len); + chunk->F.username_len = htons(chunk->F.username_len); + chunk->F.listener_port = htons(chunk->F.listener_port); header.chunk = htonl(DB_CHUNK_CLIENT); - header.length = htonl(sizeof(struct PF_client)+id_len); + header.length = htonl(sizeof(struct PF_client)+id_len+username_len); write_e(db_fptr, &header, sizeof(struct PF_header)); write_e(db_fptr, &chunk->F, sizeof(struct PF_client)); write_e(db_fptr, chunk->client_id, id_len); + if(username_len > 0){ + write_e(db_fptr, chunk->username, username_len); + } return MOSQ_ERR_SUCCESS; error: @@ -77,7 +83,7 @@ error: } -int persist__chunk_client_msg_write_v5(FILE *db_fptr, struct P_client_msg *chunk) +int persist__chunk_client_msg_write_v6(FILE *db_fptr, struct P_client_msg *chunk) { struct PF_header header; struct mosquitto__packet prop_packet; @@ -123,7 +129,7 @@ error: } -int persist__chunk_message_store_write_v5(FILE *db_fptr, struct P_msg_store *chunk) +int persist__chunk_message_store_write_v6(FILE *db_fptr, struct P_msg_store *chunk) { struct PF_header header; uint32_t payloadlen = chunk->F.payloadlen; @@ -188,7 +194,7 @@ error: } -int persist__chunk_retain_write_v5(FILE *db_fptr, struct P_retain *chunk) +int persist__chunk_retain_write_v6(FILE *db_fptr, struct P_retain *chunk) { struct PF_header header; @@ -205,7 +211,7 @@ error: } -int persist__chunk_sub_write_v5(FILE *db_fptr, struct P_sub *chunk) +int persist__chunk_sub_write_v6(FILE *db_fptr, struct P_sub *chunk) { struct PF_header header; uint16_t id_len = chunk->F.id_len; diff --git a/src/security.c b/src/security.c index 7779ca55..20b30627 100644 --- a/src/security.c +++ b/src/security.c @@ -33,7 +33,7 @@ void LIB_ERROR(void) #ifdef WIN32 char *buf; FormatMessage(FORMAT_MESSAGE_ALLOCATE_BUFFER | FORMAT_MESSAGE_FROM_STRING, - NULL, GetLastError(), LANG_NEUTRAL, &buf, 0, NULL); + NULL, GetLastError(), LANG_NEUTRAL, (LPTSTR)&buf, 0, NULL); log__printf(NULL, MOSQ_LOG_ERR, "Load error: %s", buf); LocalFree(buf); #else diff --git a/src/service.c b/src/service.c index 05bee0c5..199afe32 100644 --- a/src/service.c +++ b/src/service.c @@ -32,7 +32,7 @@ static void print_error(void) char *buf; FormatMessage(FORMAT_MESSAGE_ALLOCATE_BUFFER | FORMAT_MESSAGE_FROM_SYSTEM, - NULL, GetLastError(), LANG_NEUTRAL, &buf, 0, NULL); + NULL, GetLastError(), LANG_NEUTRAL, (LPTSTR)&buf, 0, NULL); fprintf(stderr, "Error: %s\n", buf); LocalFree(buf); @@ -70,7 +70,7 @@ void __stdcall service_main(DWORD dwArgc, LPTSTR *lpszArgv) service_handle = RegisterServiceCtrlHandler("mosquitto", service_handler); if(service_handle){ - memset(conf_path, 0, MAX_PATH + 20); + memset(conf_path, 0, sizeof(conf_path)); rc = GetEnvironmentVariable("MOSQUITTO_DIR", conf_path, MAX_PATH); if(!rc || rc == MAX_PATH){ service_status.dwCurrentState = SERVICE_STOPPED; @@ -103,25 +103,26 @@ void __stdcall service_main(DWORD dwArgc, LPTSTR *lpszArgv) void service_install(void) { SC_HANDLE sc_manager, svc_handle; - char exe_path[MAX_PATH + 5]; + char service_string[MAX_PATH + 20]; + char exe_path[MAX_PATH + 1]; SERVICE_DESCRIPTION svc_desc; - memset(exe_path, 0, MAX_PATH+5); + memset(exe_path, 0, sizeof(exe_path)); if(GetModuleFileName(NULL, exe_path, MAX_PATH) == MAX_PATH){ fprintf(stderr, "Error: Path too long.\n"); return; } - strcat(exe_path, " run"); + snprintf(service_string, sizeof(service_string), "\"%s\" run", exe_path); sc_manager = OpenSCManager(NULL, NULL, SC_MANAGER_CREATE_SERVICE); if(sc_manager){ svc_handle = CreateService(sc_manager, "mosquitto", "Mosquitto Broker", SERVICE_START | SERVICE_STOP | SERVICE_CHANGE_CONFIG, SERVICE_WIN32_OWN_PROCESS, SERVICE_AUTO_START, SERVICE_ERROR_NORMAL, - exe_path, NULL, NULL, NULL, NULL, NULL); + service_string, NULL, NULL, NULL, NULL, NULL); if(svc_handle){ - svc_desc.lpDescription = "MQTT v3.1.1 broker"; + svc_desc.lpDescription = "Eclipse Mosquitto MQTT v5/v3.1.1 broker"; ChangeServiceConfig2(svc_handle, SERVICE_CONFIG_DESCRIPTION, &svc_desc); CloseServiceHandle(svc_handle); }else{ diff --git a/src/sys_tree.h b/src/sys_tree.h index 275fb0a2..6b9804c4 100644 --- a/src/sys_tree.h +++ b/src/sys_tree.h @@ -54,10 +54,10 @@ extern unsigned int g_connection_count; #define G_MSGS_SENT_INC(A) #define G_PUB_MSGS_RECEIVED_INC(A) #define G_PUB_MSGS_SENT_INC(A) -#define G_MSGS_DROPPED_INC(A) -#define G_CLIENTS_EXPIRED_INC(A) -#define G_SOCKET_CONNECTIONS_INC(A) -#define G_CONNECTION_COUNT_INC(A) +#define G_MSGS_DROPPED_INC() +#define G_CLIENTS_EXPIRED_INC() +#define G_SOCKET_CONNECTIONS_INC() +#define G_CONNECTION_COUNT_INC() #endif diff --git a/src/websockets.c b/src/websockets.c index fccece30..039b0d89 100644 --- a/src/websockets.c +++ b/src/websockets.c @@ -302,7 +302,10 @@ static int callback_mqtt(struct libwebsocket_context *context, } count = libwebsocket_write(wsi, &packet->payload[packet->pos], txlen, LWS_WRITE_BINARY); if(count < 0){ - if (mosq->state == mosq_cs_disconnect_ws || mosq->state == mosq_cs_disconnecting){ + if (mosq->state == mosq_cs_disconnect_ws + || mosq->state == mosq_cs_disconnecting + || mosq->state == mosq_cs_disused){ + return -1; } return 0; @@ -313,7 +316,10 @@ static int callback_mqtt(struct libwebsocket_context *context, packet->to_process -= count; packet->pos += count; if(packet->to_process > 0){ - if (mosq->state == mosq_cs_disconnect_ws || mosq->state == mosq_cs_disconnecting){ + if (mosq->state == mosq_cs_disconnect_ws + || mosq->state == mosq_cs_disconnecting + || mosq->state == mosq_cs_disused){ + return -1; } break; @@ -340,7 +346,10 @@ static int callback_mqtt(struct libwebsocket_context *context, mosq->next_msg_out = mosquitto_time() + mosq->keepalive; } - if (mosq->state == mosq_cs_disconnect_ws || mosq->state == mosq_cs_disconnecting){ + if (mosq->state == mosq_cs_disconnect_ws + || mosq->state == mosq_cs_disconnecting + || mosq->state == mosq_cs_disused){ + return -1; } if(mosq->current_out_packet){ diff --git a/test/unit/files/persist_read/v5-cfg.test-db b/test/unit/files/persist_read/v5-cfg.test-db deleted file mode 100644 index 2449222e..00000000 Binary files a/test/unit/files/persist_read/v5-cfg.test-db and /dev/null differ diff --git a/test/unit/files/persist_read/v5-client-message-props.test-db b/test/unit/files/persist_read/v5-client-message-props.test-db deleted file mode 100644 index b34f6db8..00000000 Binary files a/test/unit/files/persist_read/v5-client-message-props.test-db and /dev/null differ diff --git a/test/unit/files/persist_read/v5-client-message.test-db b/test/unit/files/persist_read/v5-client-message.test-db deleted file mode 100644 index 1aac12c4..00000000 Binary files a/test/unit/files/persist_read/v5-client-message.test-db and /dev/null differ diff --git a/test/unit/files/persist_read/v5-message-store-props.test-db b/test/unit/files/persist_read/v5-message-store-props.test-db deleted file mode 100644 index a5c296a8..00000000 Binary files a/test/unit/files/persist_read/v5-message-store-props.test-db and /dev/null differ diff --git a/test/unit/files/persist_read/v5-message-store.test-db b/test/unit/files/persist_read/v5-message-store.test-db deleted file mode 100644 index f16b3907..00000000 Binary files a/test/unit/files/persist_read/v5-message-store.test-db and /dev/null differ diff --git a/test/unit/files/persist_read/v5-sub.test-db b/test/unit/files/persist_read/v5-sub.test-db deleted file mode 100644 index acd14379..00000000 Binary files a/test/unit/files/persist_read/v5-sub.test-db and /dev/null differ diff --git a/test/unit/files/persist_read/v6-cfg.test-db b/test/unit/files/persist_read/v6-cfg.test-db new file mode 100644 index 00000000..bf8151fd Binary files /dev/null and b/test/unit/files/persist_read/v6-cfg.test-db differ diff --git a/test/unit/files/persist_read/v6-client-message-props.test-db b/test/unit/files/persist_read/v6-client-message-props.test-db new file mode 100644 index 00000000..7c7a785c Binary files /dev/null and b/test/unit/files/persist_read/v6-client-message-props.test-db differ diff --git a/test/unit/files/persist_read/v6-client-message.test-db b/test/unit/files/persist_read/v6-client-message.test-db new file mode 100644 index 00000000..2010de5f Binary files /dev/null and b/test/unit/files/persist_read/v6-client-message.test-db differ diff --git a/test/unit/files/persist_read/v6-client.test-db b/test/unit/files/persist_read/v6-client.test-db new file mode 100644 index 00000000..c6be0dba Binary files /dev/null and b/test/unit/files/persist_read/v6-client.test-db differ diff --git a/test/unit/files/persist_read/v6-message-store-props.test-db b/test/unit/files/persist_read/v6-message-store-props.test-db new file mode 100644 index 00000000..982d86fe Binary files /dev/null and b/test/unit/files/persist_read/v6-message-store-props.test-db differ diff --git a/test/unit/files/persist_read/v6-message-store.test-db b/test/unit/files/persist_read/v6-message-store.test-db new file mode 100644 index 00000000..a76b8e25 Binary files /dev/null and b/test/unit/files/persist_read/v6-message-store.test-db differ diff --git a/test/unit/files/persist_read/v5-retain.test-db b/test/unit/files/persist_read/v6-retain.test-db similarity index 51% rename from test/unit/files/persist_read/v5-retain.test-db rename to test/unit/files/persist_read/v6-retain.test-db index f9d8a4f5..e515431c 100644 Binary files a/test/unit/files/persist_read/v5-retain.test-db and b/test/unit/files/persist_read/v6-retain.test-db differ diff --git a/test/unit/files/persist_read/v6-sub.test-db b/test/unit/files/persist_read/v6-sub.test-db new file mode 100644 index 00000000..f1f6663a Binary files /dev/null and b/test/unit/files/persist_read/v6-sub.test-db differ diff --git a/test/unit/files/persist_write/empty.test-db b/test/unit/files/persist_write/empty.test-db index 26169f32..eba5ad85 100644 Binary files a/test/unit/files/persist_write/empty.test-db and b/test/unit/files/persist_write/empty.test-db differ diff --git a/test/unit/files/persist_write/v5-message-store-no-ref.test-db b/test/unit/files/persist_write/v5-message-store-no-ref.test-db deleted file mode 100644 index 2449222e..00000000 Binary files a/test/unit/files/persist_write/v5-message-store-no-ref.test-db and /dev/null differ diff --git a/test/unit/files/persist_write/v6-message-store-no-ref.test-db b/test/unit/files/persist_write/v6-message-store-no-ref.test-db new file mode 100644 index 00000000..bf8151fd Binary files /dev/null and b/test/unit/files/persist_write/v6-message-store-no-ref.test-db differ diff --git a/test/unit/persist_read_test.c b/test/unit/persist_read_test.c index d1c7e786..84c36c03 100644 --- a/test/unit/persist_read_test.c +++ b/test/unit/persist_read_test.c @@ -409,7 +409,7 @@ static void TEST_v4_message_store(void) } } -static void TEST_v5_config_ok(void) +static void TEST_v6_config_ok(void) { struct mosquitto_db db; struct mosquitto__config config; @@ -420,7 +420,7 @@ static void TEST_v5_config_ok(void) db.config = &config; config.persistence = true; - config.persistence_filepath = "files/persist_read/v5-cfg.test-db"; + config.persistence_filepath = "files/persist_read/v6-cfg.test-db"; rc = persist__restore(&db); CU_ASSERT_EQUAL(rc, MOSQ_ERR_SUCCESS); @@ -466,7 +466,7 @@ static void TEST_v5_bad_chunk(void) } -static void TEST_v5_message_store(void) +static void TEST_v6_message_store(void) { struct mosquitto_db db; struct mosquitto__config config; @@ -477,7 +477,7 @@ static void TEST_v5_message_store(void) db.config = &config; config.persistence = true; - config.persistence_filepath = "files/persist_read/v5-message-store.test-db"; + config.persistence_filepath = "files/persist_read/v6-message-store.test-db"; rc = persist__restore(&db); CU_ASSERT_EQUAL(rc, MOSQ_ERR_SUCCESS); @@ -501,7 +501,7 @@ static void TEST_v5_message_store(void) } -static void TEST_v5_message_store_props(void) +static void TEST_v6_message_store_props(void) { struct mosquitto_db db; struct mosquitto__config config; @@ -518,7 +518,7 @@ static void TEST_v5_message_store_props(void) config.listener_count = 1; config.persistence = true; - config.persistence_filepath = "files/persist_read/v5-message-store-props.test-db"; + config.persistence_filepath = "files/persist_read/v6-message-store-props.test-db"; rc = persist__restore(&db); CU_ASSERT_EQUAL(rc, MOSQ_ERR_SUCCESS); @@ -573,7 +573,43 @@ static void TEST_v5_client(void) } } -static void TEST_v5_client_message(void) +static void TEST_v6_client(void) +{ + struct mosquitto_db db; + struct mosquitto__config config; + struct mosquitto *context; + struct mosquitto__listener listener; + int rc; + + memset(&db, 0, sizeof(struct mosquitto_db)); + memset(&config, 0, sizeof(struct mosquitto__config)); + memset(&listener, 0, sizeof(struct mosquitto__listener)); + db.config = &config; + + listener.port = 1883; + config.per_listener_settings = true; + config.listeners = &listener; + config.listener_count = 1; + config.persistence = true; + config.persistence_filepath = "files/persist_read/v6-client.test-db"; + + rc = persist__restore(&db); + CU_ASSERT_EQUAL(rc, MOSQ_ERR_SUCCESS); + + CU_ASSERT_PTR_NOT_NULL(db.contexts_by_id); + HASH_FIND(hh_id, db.contexts_by_id, "client-id", strlen("client-id"), context); + CU_ASSERT_PTR_NOT_NULL(context); + if(context){ + CU_ASSERT_PTR_NULL(context->msgs_in.inflight); + CU_ASSERT_PTR_NULL(context->msgs_out.inflight); + CU_ASSERT_EQUAL(context->last_mid, 0x5287); + CU_ASSERT_EQUAL(context->listener, &listener); + CU_ASSERT_PTR_NOT_NULL(context->username); + CU_ASSERT_STRING_EQUAL(context->username, "usrname"); + } +} + +static void TEST_v6_client_message(void) { struct mosquitto_db db; struct mosquitto__config config; @@ -585,7 +621,7 @@ static void TEST_v5_client_message(void) db.config = &config; config.persistence = true; - config.persistence_filepath = "files/persist_read/v5-client-message.test-db"; + config.persistence_filepath = "files/persist_read/v6-client-message.test-db"; rc = persist__restore(&db); CU_ASSERT_EQUAL(rc, MOSQ_ERR_SUCCESS); @@ -622,7 +658,7 @@ static void TEST_v5_client_message(void) } } -static void TEST_v5_client_message_props(void) +static void TEST_v6_client_message_props(void) { struct mosquitto_db db; struct mosquitto__config config; @@ -634,7 +670,7 @@ static void TEST_v5_client_message_props(void) db.config = &config; config.persistence = true; - config.persistence_filepath = "files/persist_read/v5-client-message-props.test-db"; + config.persistence_filepath = "files/persist_read/v6-client-message-props.test-db"; rc = persist__restore(&db); CU_ASSERT_EQUAL(rc, MOSQ_ERR_SUCCESS); @@ -675,7 +711,7 @@ static void TEST_v5_client_message_props(void) } } -static void TEST_v5_retain(void) +static void TEST_v6_retain(void) { struct mosquitto_db db; struct mosquitto__config config; @@ -688,7 +724,7 @@ static void TEST_v5_retain(void) db.config = &config; config.persistence = true; - config.persistence_filepath = "files/persist_read/v5-retain.test-db"; + config.persistence_filepath = "files/persist_read/v6-retain.test-db"; rc = persist__restore(&db); CU_ASSERT_EQUAL(rc, MOSQ_ERR_SUCCESS); @@ -711,7 +747,7 @@ static void TEST_v5_retain(void) CU_ASSERT_EQUAL(last_retained, 0x54); } -static void TEST_v5_sub(void) +static void TEST_v6_sub(void) { struct mosquitto_db db; struct mosquitto__config config; @@ -726,7 +762,7 @@ static void TEST_v5_sub(void) db.config = &config; config.persistence = true; - config.persistence_filepath = "files/persist_read/v5-sub.test-db"; + config.persistence_filepath = "files/persist_read/v6-sub.test-db"; rc = persist__restore(&db); CU_ASSERT_EQUAL(rc, MOSQ_ERR_SUCCESS); @@ -775,16 +811,17 @@ int init_persist_read_tests(void) || !CU_add_test(test_suite, "v3 sub", TEST_v3_sub) || !CU_add_test(test_suite, "v4 config ok", TEST_v4_config_ok) || !CU_add_test(test_suite, "v4 message store", TEST_v4_message_store) - || !CU_add_test(test_suite, "v5 config ok", TEST_v5_config_ok) + || !CU_add_test(test_suite, "v5 client", TEST_v5_client) || !CU_add_test(test_suite, "v5 config bad truncated", TEST_v5_config_truncated) || !CU_add_test(test_suite, "v5 bad chunk", TEST_v5_bad_chunk) - || !CU_add_test(test_suite, "v5 message store", TEST_v5_message_store) - || !CU_add_test(test_suite, "v5 message store+props", TEST_v5_message_store_props) - || !CU_add_test(test_suite, "v5 client", TEST_v5_client) - || !CU_add_test(test_suite, "v5 client message", TEST_v5_client_message) - || !CU_add_test(test_suite, "v5 client message+props", TEST_v5_client_message_props) - || !CU_add_test(test_suite, "v5 retain", TEST_v5_retain) - || !CU_add_test(test_suite, "v5 sub", TEST_v5_sub) + || !CU_add_test(test_suite, "v6 config ok", TEST_v6_config_ok) + || !CU_add_test(test_suite, "v6 message store", TEST_v6_message_store) + || !CU_add_test(test_suite, "v6 message store+props", TEST_v6_message_store_props) + || !CU_add_test(test_suite, "v6 client", TEST_v6_client) + || !CU_add_test(test_suite, "v6 client message", TEST_v6_client_message) + || !CU_add_test(test_suite, "v6 client message+props", TEST_v6_client_message_props) + || !CU_add_test(test_suite, "v6 retain", TEST_v6_retain) + || !CU_add_test(test_suite, "v6 sub", TEST_v6_sub) ){ printf("Error adding persist CUnit tests.\n"); diff --git a/test/unit/persist_write_test.c b/test/unit/persist_write_test.c index b6043761..16e422de 100644 --- a/test/unit/persist_write_test.c +++ b/test/unit/persist_write_test.c @@ -110,7 +110,7 @@ static void TEST_empty_file(void) } -static void TEST_v5_config_ok(void) +static void TEST_v6_config_ok(void) { struct mosquitto_db db; struct mosquitto__config config; @@ -121,20 +121,20 @@ static void TEST_v5_config_ok(void) db.config = &config; config.persistence = true; - config.persistence_filepath = "files/persist_read/v5-cfg.test-db"; + config.persistence_filepath = "files/persist_read/v6-cfg.test-db"; rc = persist__restore(&db); CU_ASSERT_EQUAL(rc, MOSQ_ERR_SUCCESS); - config.persistence_filepath = "v5-cfg.db"; + config.persistence_filepath = "v6-cfg.db"; rc = persist__backup(&db, true); CU_ASSERT_EQUAL(rc, MOSQ_ERR_SUCCESS); - CU_ASSERT_EQUAL(0, file_diff("files/persist_read/v5-cfg.test-db", "v5-cfg.db")); - unlink("v5-cfg.db"); + CU_ASSERT_EQUAL(0, file_diff("files/persist_read/v6-cfg.test-db", "v6-cfg.db")); + unlink("v6-cfg.db"); } -static void TEST_v5_message_store_no_ref(void) +static void TEST_v6_message_store_no_ref(void) { struct mosquitto_db db; struct mosquitto__config config; @@ -145,20 +145,20 @@ static void TEST_v5_message_store_no_ref(void) db.config = &config; config.persistence = true; - config.persistence_filepath = "files/persist_read/v5-message-store.test-db"; + config.persistence_filepath = "files/persist_read/v6-message-store.test-db"; rc = persist__restore(&db); CU_ASSERT_EQUAL(rc, MOSQ_ERR_SUCCESS); - config.persistence_filepath = "v5-message-store-no-ref.db"; + config.persistence_filepath = "v6-message-store-no-ref.db"; rc = persist__backup(&db, true); CU_ASSERT_EQUAL(rc, MOSQ_ERR_SUCCESS); - CU_ASSERT_EQUAL(0, file_diff("files/persist_write/v5-message-store-no-ref.test-db", "v5-message-store-no-ref.db")); - unlink("v5-message-store-no-ref.db"); + CU_ASSERT_EQUAL(0, file_diff("files/persist_write/v6-message-store-no-ref.test-db", "v6-message-store-no-ref.db")); + unlink("v6-message-store-no-ref.db"); } -static void TEST_v5_message_store_props(void) +static void TEST_v6_message_store_props(void) { struct mosquitto_db db; struct mosquitto__config config; @@ -170,48 +170,25 @@ static void TEST_v5_message_store_props(void) memset(&listener, 0, sizeof(struct mosquitto__listener)); db.config = &config; listener.port = 1883; + config.per_listener_settings = true; config.listeners = &listener; config.listener_count = 1; config.persistence = true; - config.persistence_filepath = "files/persist_read/v5-message-store-props.test-db"; + config.persistence_filepath = "files/persist_read/v6-message-store-props.test-db"; rc = persist__restore(&db); CU_ASSERT_EQUAL(rc, MOSQ_ERR_SUCCESS); - config.persistence_filepath = "v5-message-store-props.db"; + config.persistence_filepath = "v6-message-store-props.db"; rc = persist__backup(&db, true); CU_ASSERT_EQUAL(rc, MOSQ_ERR_SUCCESS); - CU_ASSERT_EQUAL(0, file_diff("files/persist_read/v5-message-store-props.test-db", "v5-message-store-props.db")); - unlink("v5-message-store-props.db"); + CU_ASSERT_EQUAL(0, file_diff("files/persist_read/v6-message-store-props.test-db", "v6-message-store-props.db")); + unlink("v6-message-store-props.db"); } -static void TEST_v5_client(void) -{ - struct mosquitto_db db; - struct mosquitto__config config; - int rc; - - memset(&db, 0, sizeof(struct mosquitto_db)); - memset(&config, 0, sizeof(struct mosquitto__config)); - db.config = &config; - - config.persistence = true; - config.persistence_filepath = "files/persist_read/v5-client.test-db"; - rc = persist__restore(&db); - CU_ASSERT_EQUAL(rc, MOSQ_ERR_SUCCESS); - - config.persistence_filepath = "v5-client.db"; - rc = persist__backup(&db, true); - CU_ASSERT_EQUAL(rc, MOSQ_ERR_SUCCESS); - - CU_ASSERT_EQUAL(0, file_diff("files/persist_read/v5-client.test-db", "v5-client.db")); - unlink("v5-client.db"); -} - - -static void TEST_v5_client_message(void) +static void TEST_v6_client(void) { struct mosquitto_db db; struct mosquitto__config config; @@ -223,24 +200,25 @@ static void TEST_v5_client_message(void) memset(&listener, 0, sizeof(struct mosquitto__listener)); db.config = &config; listener.port = 1883; + config.per_listener_settings = true; config.listeners = &listener; config.listener_count = 1; config.persistence = true; - config.persistence_filepath = "files/persist_read/v5-client-message.test-db"; + config.persistence_filepath = "files/persist_read/v6-client.test-db"; rc = persist__restore(&db); CU_ASSERT_EQUAL(rc, MOSQ_ERR_SUCCESS); - config.persistence_filepath = "v5-client-message.db"; + config.persistence_filepath = "v6-client.db"; rc = persist__backup(&db, true); CU_ASSERT_EQUAL(rc, MOSQ_ERR_SUCCESS); - CU_ASSERT_EQUAL(0, file_diff("files/persist_read/v5-client-message.test-db", "v5-client-message.db")); - unlink("v5-client-message.db"); + CU_ASSERT_EQUAL(0, file_diff("files/persist_read/v6-client.test-db", "v6-client.db")); + unlink("v6-client.db"); } -static void TEST_v5_client_message_props(void) +static void TEST_v6_client_message(void) { struct mosquitto_db db; struct mosquitto__config config; @@ -252,11 +230,42 @@ static void TEST_v5_client_message_props(void) memset(&listener, 0, sizeof(struct mosquitto__listener)); db.config = &config; listener.port = 1883; + config.per_listener_settings = true; config.listeners = &listener; config.listener_count = 1; config.persistence = true; - config.persistence_filepath = "files/persist_read/v5-client-message-props.test-db"; + config.persistence_filepath = "files/persist_read/v6-client-message.test-db"; + rc = persist__restore(&db); + CU_ASSERT_EQUAL(rc, MOSQ_ERR_SUCCESS); + + config.persistence_filepath = "v6-client-message.db"; + rc = persist__backup(&db, true); + CU_ASSERT_EQUAL(rc, MOSQ_ERR_SUCCESS); + + CU_ASSERT_EQUAL(0, file_diff("files/persist_read/v6-client-message.test-db", "v6-client-message.db")); + unlink("v6-client-message.db"); +} + + +static void TEST_v6_client_message_props(void) +{ + struct mosquitto_db db; + struct mosquitto__config config; + struct mosquitto__listener listener; + int rc; + + memset(&db, 0, sizeof(struct mosquitto_db)); + memset(&config, 0, sizeof(struct mosquitto__config)); + memset(&listener, 0, sizeof(struct mosquitto__listener)); + db.config = &config; + listener.port = 1883; + config.per_listener_settings = true; + config.listeners = &listener; + config.listener_count = 1; + + config.persistence = true; + config.persistence_filepath = "files/persist_read/v6-client-message-props.test-db"; rc = persist__restore(&db); CU_ASSERT_EQUAL(rc, MOSQ_ERR_SUCCESS); @@ -268,16 +277,16 @@ static void TEST_v5_client_message_props(void) } } - config.persistence_filepath = "v5-client-message-props.db"; + config.persistence_filepath = "v6-client-message-props.db"; rc = persist__backup(&db, true); CU_ASSERT_EQUAL(rc, MOSQ_ERR_SUCCESS); - CU_ASSERT_EQUAL(0, file_diff("files/persist_read/v5-client-message-props.test-db", "v5-client-message-props.db")); - unlink("v5-client-message-props.db"); + CU_ASSERT_EQUAL(0, file_diff("files/persist_read/v6-client-message-props.test-db", "v6-client-message-props.db")); + unlink("v6-client-message-props.db"); } -static void TEST_v5_sub(void) +static void TEST_v6_sub(void) { struct mosquitto_db db; struct mosquitto__config config; @@ -289,22 +298,23 @@ static void TEST_v5_sub(void) memset(&listener, 0, sizeof(struct mosquitto__listener)); db.config = &config; listener.port = 1883; + config.per_listener_settings = true; config.listeners = &listener; config.listener_count = 1; db__open(&config, &db); config.persistence = true; - config.persistence_filepath = "files/persist_read/v5-sub.test-db"; + config.persistence_filepath = "files/persist_read/v6-sub.test-db"; rc = persist__restore(&db); CU_ASSERT_EQUAL(rc, MOSQ_ERR_SUCCESS); - config.persistence_filepath = "v5-sub.db"; + config.persistence_filepath = "v6-sub.db"; rc = persist__backup(&db, true); CU_ASSERT_EQUAL(rc, MOSQ_ERR_SUCCESS); - CU_ASSERT_EQUAL(0, file_diff("files/persist_read/v5-sub.test-db", "v5-sub.db")); - unlink("v5-sub.db"); + CU_ASSERT_EQUAL(0, file_diff("files/persist_read/v6-sub.test-db", "v6-sub.db")); + unlink("v6-sub.db"); } @@ -362,13 +372,13 @@ int main(int argc, char *argv[]) if(0 || !CU_add_test(test_suite, "Persistence disabled", TEST_persistence_disabled) || !CU_add_test(test_suite, "Empty file", TEST_empty_file) - || !CU_add_test(test_suite, "v5 config ok", TEST_v5_config_ok) - || !CU_add_test(test_suite, "v5 message store (message has no refs)", TEST_v5_message_store_no_ref) - || !CU_add_test(test_suite, "v5 message store + props", TEST_v5_message_store_props) - || !CU_add_test(test_suite, "v5 client", TEST_v5_client) - || !CU_add_test(test_suite, "v5 client message", TEST_v5_client_message) - || !CU_add_test(test_suite, "v5 client message+props", TEST_v5_client_message_props) - || !CU_add_test(test_suite, "v5 sub", TEST_v5_sub) + || !CU_add_test(test_suite, "v6 config ok", TEST_v6_config_ok) + || !CU_add_test(test_suite, "v6 message store (message has no refs)", TEST_v6_message_store_no_ref) + || !CU_add_test(test_suite, "v6 message store + props", TEST_v6_message_store_props) + || !CU_add_test(test_suite, "v6 client", TEST_v6_client) + || !CU_add_test(test_suite, "v6 client message", TEST_v6_client_message) + || !CU_add_test(test_suite, "v6 client message+props", TEST_v6_client_message_props) + || !CU_add_test(test_suite, "v6 sub", TEST_v6_sub) //|| !CU_add_test(test_suite, "v5 full", TEST_v5_full) ){