Mirrors_Framework/mosquitto
1
0
Fork 0
mirror of https://github.com/eclipse-mosquitto/mosquitto.git synced 2026-09-24 09:14:02 +08:00
Code Issues Packages Projects Releases Wiki Activity

Store username and listener of persistent clients

Browse Source 
Signed-off-by: david-beinder <david.beinder@mce.li>
This commit is contained in:
david-beinder
2020-08-07 08:32:14 +02:00
parent 618cd7006b
commit 212fd8635f
8 changed files with 147 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/db_dump/db_dump.c
+12 -12
View File
@@ -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);
}
src/db_dump/print.c
+6
View File
@@ -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);
src/mosquitto.c
+12
View File
@@ -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
src/persist.h
+28 -15
View File
@@ -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
src/persist_read.c
+35 -17
View File
@@ -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;
}
src/persist_read_v5.c
+28 -10
View File
@@ -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;
src/persist_write.c
+13 -6
View File
@@ -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;
}
src/persist_write_v5.c
+13 -7
View File
@@ -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;