From 212fd8635f5e31fcf866d64f5e2ee87243735e03 Mon Sep 17 00:00:00 2001 From: david-beinder Date: Fri, 7 Aug 2020 08:32:14 +0200 Subject: [PATCH] Store username and listener of persistent clients Signed-off-by: david-beinder --- src/db_dump/db_dump.c | 24 +++++++++---------- src/db_dump/print.c | 6 +++++ src/mosquitto.c | 12 ++++++++++ src/persist.h | 43 ++++++++++++++++++++++------------ src/persist_read.c | 52 ++++++++++++++++++++++++++++-------------- src/persist_read_v5.c | 38 ++++++++++++++++++++++-------- src/persist_write.c | 19 ++++++++++----- src/persist_write_v5.c | 20 ++++++++++------ 8 files changed, 147 insertions(+), 67 deletions(-) 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/mosquitto.c b/src/mosquitto.c index fdde843b..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 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;