mirror of
https://github.com/servalproject/serval-dna.git
synced 2025-04-13 05:43:07 +00:00
Improve MeshMS failure handling
Introduce "enum meshms_status" to replace int -1,0,1 for return values from MeshMS internal functions Add test cases to check that status 2 is returned when attempting to send a message from an identity that is not known (unlocked)
This commit is contained in:
Andrew Bettison
parent
f1ceffbc4e
commit
5dd9ea7a6b
488
meshms.c
488
meshms.c
@ -17,6 +17,7 @@ along with this program; if not, write to the Free Software
|
||||
Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
|
||||
*/
|
||||
|
||||
#define __MESHMS_INLINE
|
||||
#include <assert.h>
|
||||
#include "serval.h"
|
||||
#include "rhizome.h"
|
||||
@ -42,12 +43,12 @@ void meshms_free_conversations(struct meshms_conversations *conv)
|
||||
}
|
||||
}
|
||||
|
||||
static int get_my_conversation_bundle(const sid_t *my_sidp, rhizome_manifest *m)
|
||||
static enum meshms_status get_my_conversation_bundle(const sid_t *my_sidp, rhizome_manifest *m)
|
||||
{
|
||||
/* Find our private key */
|
||||
unsigned cn=0, in=0, kp=0;
|
||||
if (!keyring_find_sid(keyring,&cn,&in,&kp,my_sidp))
|
||||
return WHYF("SID was not found in keyring: %s", alloca_tohex_sid_t(*my_sidp));
|
||||
return MESHMS_STATUS_SID_LOCKED;
|
||||
|
||||
char seed[1024];
|
||||
snprintf(seed, sizeof(seed),
|
||||
@ -56,7 +57,7 @@ static int get_my_conversation_bundle(const sid_t *my_sidp, rhizome_manifest *m)
|
||||
->keypairs[kp]->private_key, crypto_box_curve25519xsalsa20poly1305_SECRETKEYBYTES));
|
||||
|
||||
if (rhizome_get_bundle_from_seed(m, seed) == -1)
|
||||
return -1;
|
||||
return MESHMS_STATUS_ERROR;
|
||||
|
||||
// always consider the content encrypted, we don't need to rely on the manifest itself.
|
||||
rhizome_manifest_set_crypt(m, PAYLOAD_ENCRYPTED);
|
||||
@ -76,10 +77,12 @@ static int get_my_conversation_bundle(const sid_t *my_sidp, rhizome_manifest *m)
|
||||
);
|
||||
}
|
||||
} else {
|
||||
if (strcmp(m->service, RHIZOME_SERVICE_FILE) != 0)
|
||||
return WHYF("Invalid manifest, service=%s but should be %s", m->service, RHIZOME_SERVICE_MESHMS2);
|
||||
if (strcmp(m->service, RHIZOME_SERVICE_FILE) != 0) {
|
||||
WARNF("Invalid conversations manifest, service=%s but should be %s", m->service, RHIZOME_SERVICE_FILE);
|
||||
return MESHMS_STATUS_PROTOCOL_FAULT;
|
||||
}
|
||||
}
|
||||
return 0;
|
||||
return MESHMS_STATUS_OK;
|
||||
}
|
||||
|
||||
static struct meshms_conversations *add_conv(struct meshms_conversations **conv, const sid_t *them)
|
||||
@ -96,18 +99,16 @@ static struct meshms_conversations *add_conv(struct meshms_conversations **conv,
|
||||
else
|
||||
ptr = &(*ptr)->_right;
|
||||
}
|
||||
if (!*ptr){
|
||||
*ptr = emalloc_zero(sizeof(struct meshms_conversations));
|
||||
if (*ptr == NULL && (*ptr = emalloc_zero(sizeof(struct meshms_conversations))) != NULL) {
|
||||
(*ptr)->_parent = parent;
|
||||
if (*ptr)
|
||||
(*ptr)->them = *them;
|
||||
(*ptr)->them = *them;
|
||||
}
|
||||
return *ptr;
|
||||
}
|
||||
|
||||
// find matching conversations
|
||||
// if their_sid == my_sid, return all conversations with any recipient
|
||||
static int get_database_conversations(const sid_t *my_sid, const sid_t *their_sid, struct meshms_conversations **conv)
|
||||
static enum meshms_status get_database_conversations(const sid_t *my_sid, const sid_t *their_sid, struct meshms_conversations **conv)
|
||||
{
|
||||
sqlite_retry_state retry = SQLITE_RETRY_STATE_DEFAULT;
|
||||
sqlite3_stmt *statement = sqlite_prepare_bind(&retry,
|
||||
@ -122,7 +123,7 @@ static int get_database_conversations(const sid_t *my_sid, const sid_t *their_si
|
||||
END
|
||||
);
|
||||
if (!statement)
|
||||
return -1;
|
||||
return MESHMS_STATUS_ERROR;
|
||||
if (config.debug.meshms) {
|
||||
const char *my_sid_hex = alloca_tohex_sid_t(*my_sid);
|
||||
const char *their_sid_hex = alloca_tohex_sid_t(*(their_sid ? their_sid : my_sid));
|
||||
@ -172,17 +173,21 @@ static int get_database_conversations(const sid_t *my_sid, const sid_t *their_si
|
||||
p->size = size;
|
||||
}
|
||||
sqlite3_finalize(statement);
|
||||
return 0;
|
||||
return MESHMS_STATUS_OK;
|
||||
}
|
||||
|
||||
static struct meshms_conversations * find_or_create_conv(const sid_t *my_sid, const sid_t *their_sid)
|
||||
static enum meshms_status find_or_create_conv(const sid_t *my_sid, const sid_t *their_sid, struct meshms_conversations **conv)
|
||||
{
|
||||
struct meshms_conversations *conv=NULL;
|
||||
if (meshms_conversations_list(my_sid, their_sid, &conv))
|
||||
return NULL;
|
||||
if (!conv && (conv = emalloc_zero(sizeof(struct meshms_conversations))))
|
||||
conv->them = *their_sid;
|
||||
return conv;
|
||||
enum meshms_status status;
|
||||
if (meshms_failed(status = meshms_conversations_list(my_sid, their_sid, conv)))
|
||||
return status;
|
||||
if (*conv == NULL) {
|
||||
if ((*conv = (struct meshms_conversations *) emalloc_zero(sizeof(struct meshms_conversations))) == NULL)
|
||||
return MESHMS_STATUS_ERROR;
|
||||
(*conv)->them = *their_sid;
|
||||
status = MESHMS_STATUS_UPDATED;
|
||||
}
|
||||
return status;
|
||||
}
|
||||
|
||||
static int create_ply(const sid_t *my_sid, struct meshms_conversations *conv, rhizome_manifest *m)
|
||||
@ -213,20 +218,32 @@ static int append_footer(unsigned char *buffer, char type, size_t message_len)
|
||||
return 2;
|
||||
}
|
||||
|
||||
static int ply_read_open(struct meshms_ply_read *ply, const rhizome_bid_t *bid, rhizome_manifest *m)
|
||||
static enum meshms_status ply_read_open(struct meshms_ply_read *ply, const rhizome_bid_t *bid, rhizome_manifest *m)
|
||||
{
|
||||
if (config.debug.meshms)
|
||||
DEBUGF("Opening ply %s", alloca_tohex_rhizome_bid_t(*bid));
|
||||
if (rhizome_retrieve_manifest(bid, m))
|
||||
return -1;
|
||||
switch (rhizome_retrieve_manifest(bid, m)) {
|
||||
case 0:
|
||||
break;
|
||||
case -1:
|
||||
return MESHMS_STATUS_ERROR;
|
||||
default: // bundle not found
|
||||
return MESHMS_STATUS_PROTOCOL_FAULT;
|
||||
}
|
||||
enum rhizome_payload_status pstatus = rhizome_open_decrypt_read(m, &ply->read);
|
||||
if (pstatus == RHIZOME_PAYLOAD_STATUS_NEW)
|
||||
if (config.debug.meshms)
|
||||
DEBUGF("pstatus=%d", pstatus);
|
||||
if (pstatus == RHIZOME_PAYLOAD_STATUS_NEW) {
|
||||
WARNF("Payload was not found for manifest %s, %"PRIu64, alloca_tohex_rhizome_bid_t(m->cryptoSignPublic), m->version);
|
||||
return MESHMS_STATUS_PROTOCOL_FAULT;
|
||||
}
|
||||
if (pstatus == RHIZOME_PAYLOAD_STATUS_CRYPTO_FAIL)
|
||||
return MESHMS_STATUS_SID_LOCKED;
|
||||
if (pstatus != RHIZOME_PAYLOAD_STATUS_STORED && pstatus != RHIZOME_PAYLOAD_STATUS_EMPTY)
|
||||
return -1;
|
||||
return MESHMS_STATUS_ERROR;
|
||||
assert(m->filesize != RHIZOME_SIZE_UNSET);
|
||||
ply->read.offset = ply->read.length = m->filesize;
|
||||
return 0;
|
||||
return MESHMS_STATUS_OK;
|
||||
}
|
||||
|
||||
static void ply_read_close(struct meshms_ply_read *ply)
|
||||
@ -241,22 +258,27 @@ static void ply_read_close(struct meshms_ply_read *ply)
|
||||
}
|
||||
|
||||
// read the next record from the ply (backwards)
|
||||
// returns 1 on EOF, -1 on failure
|
||||
static int ply_read_prev(struct meshms_ply_read *ply)
|
||||
// returns MESHMS_STATUS_UPDATED if the read advances to a new record, MESHMS_STATUS_OK if at the
|
||||
// end of records
|
||||
static enum meshms_status ply_read_prev(struct meshms_ply_read *ply)
|
||||
{
|
||||
ply->record_end_offset = ply->read.offset;
|
||||
unsigned char footer[2];
|
||||
if (ply->read.offset <= sizeof footer) {
|
||||
if (config.debug.meshms)
|
||||
DEBUG("EOF");
|
||||
return 1;
|
||||
return MESHMS_STATUS_OK;
|
||||
}
|
||||
ply->read.offset -= sizeof footer;
|
||||
ssize_t read = rhizome_read_buffered(&ply->read, &ply->buff, footer, sizeof footer);
|
||||
if (read == -1)
|
||||
return WHYF("rhizome_read_buffered() failed");
|
||||
if ((size_t) read != sizeof footer)
|
||||
return WHYF("Expected %zu bytes read, got %zu", (size_t) sizeof footer, (size_t) read);
|
||||
if (read == -1) {
|
||||
WHYF("rhizome_read_buffered() failed");
|
||||
return MESHMS_STATUS_ERROR;
|
||||
}
|
||||
if ((size_t) read != sizeof footer) {
|
||||
WHYF("Expected %zu bytes read, got %zu", (size_t) sizeof footer, (size_t) read);
|
||||
return MESHMS_STATUS_PROTOCOL_FAULT;
|
||||
}
|
||||
// (rhizome_read automatically advances the offset by the number of bytes read)
|
||||
ply->record_length=read_uint16(footer);
|
||||
ply->type = ply->record_length & 0xF;
|
||||
@ -267,7 +289,7 @@ static int ply_read_prev(struct meshms_ply_read *ply)
|
||||
if (ply->record_length + sizeof footer > ply->read.offset){
|
||||
if (config.debug.meshms)
|
||||
DEBUGF("EOF");
|
||||
return 1;
|
||||
return MESHMS_STATUS_OK;
|
||||
}
|
||||
ply->read.offset -= ply->record_length + sizeof(footer);
|
||||
uint64_t record_start = ply->read.offset;
|
||||
@ -275,122 +297,134 @@ static int ply_read_prev(struct meshms_ply_read *ply)
|
||||
ply->record_size = ply->record_length;
|
||||
unsigned char *b = erealloc(ply->record, ply->record_size);
|
||||
if (!b)
|
||||
return -1;
|
||||
return MESHMS_STATUS_ERROR;
|
||||
ply->record = b;
|
||||
}
|
||||
read = rhizome_read_buffered(&ply->read, &ply->buff, ply->record, ply->record_length);
|
||||
if (read == -1)
|
||||
if (read == -1) {
|
||||
return WHYF("rhizome_read_buffered() failed");
|
||||
if ((size_t) read != ply->record_length)
|
||||
return WHYF("Expected %u bytes read, got %zu", ply->record_length, (size_t) read);
|
||||
return MESHMS_STATUS_ERROR;
|
||||
}
|
||||
if ((size_t) read != ply->record_length) {
|
||||
WHYF("Expected %u bytes read, got %zu", ply->record_length, (size_t) read);
|
||||
return MESHMS_STATUS_PROTOCOL_FAULT;
|
||||
}
|
||||
ply->read.offset = record_start;
|
||||
return 0;
|
||||
return MESHMS_STATUS_UPDATED;
|
||||
}
|
||||
|
||||
// keep reading past messages until you find this type.
|
||||
static int ply_find_prev(struct meshms_ply_read *ply, char type)
|
||||
static enum meshms_status ply_find_prev(struct meshms_ply_read *ply, char type)
|
||||
{
|
||||
int ret;
|
||||
while ((ret = ply_read_prev(ply)) == 0 && ply->type != type)
|
||||
enum meshms_status status;
|
||||
while ((status = ply_read_prev(ply)) == MESHMS_STATUS_UPDATED && ply->type != type)
|
||||
;
|
||||
return ret;
|
||||
return status;
|
||||
}
|
||||
|
||||
static int append_meshms_buffer(const sid_t *my_sid, struct meshms_conversations *conv, unsigned char *buffer, int len)
|
||||
static enum meshms_status append_meshms_buffer(const sid_t *my_sid, struct meshms_conversations *conv, unsigned char *buffer, int len)
|
||||
{
|
||||
int ret=-1;
|
||||
enum meshms_status status = MESHMS_STATUS_ERROR;
|
||||
rhizome_manifest *mout = NULL;
|
||||
rhizome_manifest *m = rhizome_new_manifest();
|
||||
if (!m)
|
||||
goto end;
|
||||
|
||||
if (conv->found_my_ply){
|
||||
if (rhizome_retrieve_manifest(&conv->my_ply.bundle_id, m))
|
||||
goto end;
|
||||
switch (rhizome_retrieve_manifest(&conv->my_ply.bundle_id, m)) {
|
||||
case 0:
|
||||
break;
|
||||
case -1:
|
||||
status = MESHMS_STATUS_ERROR;
|
||||
goto end;
|
||||
default:
|
||||
status = MESHMS_STATUS_PROTOCOL_FAULT;
|
||||
goto end;
|
||||
}
|
||||
rhizome_authenticate_author(m);
|
||||
if (!m->haveSecret || m->authorship != AUTHOR_AUTHENTIC)
|
||||
goto end;
|
||||
}else{
|
||||
if (create_ply(my_sid, conv, m))
|
||||
if (!m->haveSecret || m->authorship != AUTHOR_AUTHENTIC) {
|
||||
status = MESHMS_STATUS_PROTOCOL_FAULT;
|
||||
goto end;
|
||||
}
|
||||
} else if (create_ply(my_sid, conv, m) == -1) {
|
||||
status = MESHMS_STATUS_ERROR;
|
||||
goto end;
|
||||
}
|
||||
assert(m->haveSecret);
|
||||
assert(m->authorship == AUTHOR_AUTHENTIC);
|
||||
|
||||
enum rhizome_payload_status pstatus = rhizome_append_journal_buffer(m, 0, buffer, len);
|
||||
if (pstatus != RHIZOME_PAYLOAD_STATUS_NEW)
|
||||
if (pstatus != RHIZOME_PAYLOAD_STATUS_NEW) {
|
||||
status = MESHMS_STATUS_ERROR;
|
||||
goto end;
|
||||
|
||||
enum rhizome_bundle_status status = rhizome_manifest_finalise(m, &mout, 1);
|
||||
switch (status) {
|
||||
}
|
||||
enum rhizome_bundle_status bstatus = rhizome_manifest_finalise(m, &mout, 1);
|
||||
if (config.debug.meshms)
|
||||
DEBUGF("bstatus=%d", bstatus);
|
||||
switch (bstatus) {
|
||||
case RHIZOME_BUNDLE_STATUS_ERROR:
|
||||
// error is already logged
|
||||
// error has already been logged
|
||||
status = MESHMS_STATUS_ERROR;
|
||||
break;
|
||||
case RHIZOME_BUNDLE_STATUS_NEW:
|
||||
ret = 0;
|
||||
status = MESHMS_STATUS_UPDATED;
|
||||
break;
|
||||
case RHIZOME_BUNDLE_STATUS_SAME:
|
||||
case RHIZOME_BUNDLE_STATUS_DUPLICATE:
|
||||
case RHIZOME_BUNDLE_STATUS_OLD:
|
||||
status = MESHMS_STATUS_PROTOCOL_FAULT;
|
||||
WHYF("MeshMS ply manifest (version=%"PRIu64") gazumped by Rhizome store (version=%"PRIu64")",
|
||||
m->version, mout->version);
|
||||
break;
|
||||
case RHIZOME_BUNDLE_STATUS_INCONSISTENT:
|
||||
status = MESHMS_STATUS_PROTOCOL_FAULT;
|
||||
WHYF("MeshMS ply manifest not consistent with payload");
|
||||
break;
|
||||
case RHIZOME_BUNDLE_STATUS_FAKE:
|
||||
status = MESHMS_STATUS_PROTOCOL_FAULT;
|
||||
WHYF("MeshMS ply manifest is not signed");
|
||||
break;
|
||||
case RHIZOME_BUNDLE_STATUS_INVALID:
|
||||
status = MESHMS_STATUS_PROTOCOL_FAULT;
|
||||
WHYF("MeshMS ply manifest is invalid");
|
||||
break;
|
||||
default:
|
||||
FATALF("status=%d", status);
|
||||
FATALF("bstatus=%d", bstatus);
|
||||
}
|
||||
end:
|
||||
if (mout && mout!=m)
|
||||
rhizome_manifest_free(mout);
|
||||
if (m)
|
||||
rhizome_manifest_free(m);
|
||||
return ret;
|
||||
return status;
|
||||
}
|
||||
|
||||
// update if any conversations are unread or need to be acked.
|
||||
// return -1 for failure, 1 if the conversation index needs to be saved.
|
||||
static int update_conversation(const sid_t *my_sid, struct meshms_conversations *conv)
|
||||
// return MESHMS_STATUS_UPDATED if the conversation index needs to be saved.
|
||||
static enum meshms_status update_conversation(const sid_t *my_sid, struct meshms_conversations *conv)
|
||||
{
|
||||
if (config.debug.meshms)
|
||||
DEBUG("Checking if conversation needs to be acked");
|
||||
|
||||
// Nothing to be done if they have never sent us anything
|
||||
if (!conv->found_their_ply)
|
||||
return 0;
|
||||
return MESHMS_STATUS_OK;
|
||||
|
||||
rhizome_manifest *m_ours = NULL;
|
||||
rhizome_manifest *m_theirs = rhizome_new_manifest();
|
||||
if (!m_theirs)
|
||||
return -1;
|
||||
return MESHMS_STATUS_ERROR;
|
||||
|
||||
struct meshms_ply_read ply;
|
||||
bzero(&ply, sizeof(ply));
|
||||
int ret=-1;
|
||||
|
||||
enum meshms_status status = MESHMS_STATUS_ERROR;
|
||||
if (config.debug.meshms)
|
||||
DEBUG("Locating their last message");
|
||||
|
||||
if (ply_read_open(&ply, &conv->their_ply.bundle_id, m_theirs))
|
||||
if (meshms_failed(status = ply_read_open(&ply, &conv->their_ply.bundle_id, m_theirs)))
|
||||
goto end;
|
||||
ret = ply_find_prev(&ply, MESHMS_BLOCK_TYPE_MESSAGE);
|
||||
if (ret!=0){
|
||||
// no messages indicates that we didn't do anthing
|
||||
if (ret>0)
|
||||
ret=0;
|
||||
if (meshms_failed(status = ply_find_prev(&ply, MESHMS_BLOCK_TYPE_MESSAGE)))
|
||||
goto end;
|
||||
}
|
||||
|
||||
if (conv->their_last_message == ply.record_end_offset){
|
||||
// nothing has changed since last time
|
||||
ret=0;
|
||||
status = MESHMS_STATUS_OK;
|
||||
goto end;
|
||||
}
|
||||
|
||||
@ -407,17 +441,18 @@ static int update_conversation(const sid_t *my_sid, struct meshms_conversations
|
||||
DEBUG("Locating our previous ack");
|
||||
|
||||
m_ours = rhizome_new_manifest();
|
||||
if (!m_ours)
|
||||
if (!m_ours) {
|
||||
status = MESHMS_STATUS_ERROR;
|
||||
goto end;
|
||||
if (ply_read_open(&ply, &conv->my_ply.bundle_id, m_ours))
|
||||
}
|
||||
if (meshms_failed(status = ply_read_open(&ply, &conv->my_ply.bundle_id, m_ours)))
|
||||
goto end;
|
||||
ret = ply_find_prev(&ply, MESHMS_BLOCK_TYPE_ACK);
|
||||
if (ret == -1)
|
||||
if (meshms_failed(status = ply_find_prev(&ply, MESHMS_BLOCK_TYPE_ACK)))
|
||||
goto end;
|
||||
|
||||
if (ret==0){
|
||||
if (status == MESHMS_STATUS_UPDATED) {
|
||||
if (unpack_uint(ply.record, ply.record_length, &previous_ack) == -1)
|
||||
previous_ack=0;
|
||||
status = MESHMS_STATUS_OK;
|
||||
}
|
||||
if (config.debug.meshms)
|
||||
DEBUGF("Previous ack is %"PRId64, previous_ack);
|
||||
@ -425,89 +460,98 @@ static int update_conversation(const sid_t *my_sid, struct meshms_conversations
|
||||
}else{
|
||||
if (config.debug.meshms)
|
||||
DEBUGF("No outgoing ply");
|
||||
status = MESHMS_STATUS_PROTOCOL_FAULT;
|
||||
}
|
||||
|
||||
if (previous_ack >= conv->their_last_message){
|
||||
// their last message has already been acked
|
||||
ret=1;
|
||||
status = MESHMS_STATUS_UPDATED;
|
||||
goto end;
|
||||
}
|
||||
|
||||
|
||||
// append an ack for their message
|
||||
if (config.debug.meshms)
|
||||
DEBUGF("Creating ACK for %"PRId64" - %"PRId64, previous_ack, conv->their_last_message);
|
||||
|
||||
unsigned char buffer[24];
|
||||
int ofs=0;
|
||||
ofs+=pack_uint(&buffer[ofs], conv->their_last_message);
|
||||
if (previous_ack)
|
||||
ofs+=pack_uint(&buffer[ofs], conv->their_last_message - previous_ack);
|
||||
ofs+=append_footer(buffer+ofs, MESHMS_BLOCK_TYPE_ACK, ofs);
|
||||
ret = append_meshms_buffer(my_sid, conv, buffer, ofs);
|
||||
|
||||
status = append_meshms_buffer(my_sid, conv, buffer, ofs);
|
||||
if (config.debug.meshms)
|
||||
DEBUGF("status=%d", status);
|
||||
end:
|
||||
ply_read_close(&ply);
|
||||
if (m_ours)
|
||||
rhizome_manifest_free(m_ours);
|
||||
if (m_theirs)
|
||||
rhizome_manifest_free(m_theirs);
|
||||
|
||||
// if it's all good, remember the size of their ply at the time we examined it.
|
||||
if (ret>=0)
|
||||
if (!meshms_failed(status))
|
||||
conv->their_size = conv->their_ply.size;
|
||||
|
||||
return ret;
|
||||
return status;
|
||||
}
|
||||
|
||||
// update conversations, and return 1 if the conversation index should be saved
|
||||
static int update_conversations(const sid_t *my_sid, struct meshms_conversations *conv)
|
||||
// update conversations, and return MESHMS_STATUS_UPDATED if the conversation index should be saved
|
||||
static enum meshms_status update_conversations(const sid_t *my_sid, struct meshms_conversations *conv)
|
||||
{
|
||||
if (!conv)
|
||||
return 0;
|
||||
int ret = 0;
|
||||
if (update_conversations(my_sid, conv->_left))
|
||||
ret=1;
|
||||
|
||||
if (conv->their_size != conv->their_ply.size){
|
||||
if (update_conversation(my_sid, conv)>0)
|
||||
ret=1;
|
||||
enum meshms_status rstatus = MESHMS_STATUS_OK;
|
||||
if (conv) {
|
||||
enum meshms_status status;
|
||||
if (meshms_failed(status = update_conversations(my_sid, conv->_left)))
|
||||
return status;
|
||||
if (status == MESHMS_STATUS_UPDATED)
|
||||
rstatus = MESHMS_STATUS_UPDATED;
|
||||
if (conv->their_size != conv->their_ply.size) {
|
||||
if (meshms_failed(status = update_conversation(my_sid, conv)))
|
||||
return status;
|
||||
if (status == MESHMS_STATUS_UPDATED)
|
||||
rstatus = MESHMS_STATUS_UPDATED;
|
||||
}
|
||||
if (meshms_failed(status = update_conversations(my_sid, conv->_right)))
|
||||
return status;
|
||||
if (status == MESHMS_STATUS_UPDATED)
|
||||
rstatus = MESHMS_STATUS_UPDATED;
|
||||
}
|
||||
|
||||
if (update_conversations(my_sid, conv->_right))
|
||||
ret=1;
|
||||
|
||||
return ret;
|
||||
return rstatus;
|
||||
}
|
||||
|
||||
// read our cached conversation list from our rhizome payload
|
||||
// if we can't load the existing data correctly, just ignore it.
|
||||
static int read_known_conversations(rhizome_manifest *m, const sid_t *their_sid, struct meshms_conversations **conv)
|
||||
static enum meshms_status read_known_conversations(rhizome_manifest *m, const sid_t *their_sid, struct meshms_conversations **conv)
|
||||
{
|
||||
if (m->haveSecret==NEW_BUNDLE_ID)
|
||||
return 0;
|
||||
|
||||
return MESHMS_STATUS_OK;
|
||||
|
||||
struct rhizome_read read;
|
||||
bzero(&read, sizeof(read));
|
||||
struct rhizome_read_buffer buff;
|
||||
bzero(&buff, sizeof(buff));
|
||||
|
||||
int ret = -1;
|
||||
|
||||
enum meshms_status status = MESHMS_STATUS_ERROR;
|
||||
enum rhizome_payload_status pstatus = rhizome_open_decrypt_read(m, &read);
|
||||
if (pstatus != RHIZOME_PAYLOAD_STATUS_STORED)
|
||||
if (pstatus == RHIZOME_PAYLOAD_STATUS_NEW) {
|
||||
WARNF("Payload was not found for manifest %s, %"PRIu64, alloca_tohex_rhizome_bid_t(m->cryptoSignPublic), m->version);
|
||||
goto fault;
|
||||
}
|
||||
if (pstatus != RHIZOME_PAYLOAD_STATUS_STORED && pstatus != RHIZOME_PAYLOAD_STATUS_EMPTY)
|
||||
goto end;
|
||||
|
||||
|
||||
unsigned char version=0xFF;
|
||||
ssize_t r = rhizome_read_buffered(&read, &buff, &version, 1);
|
||||
if (r == -1)
|
||||
goto end;
|
||||
if (version != 1) {
|
||||
WARNF("Expected version 1 (got 0x%02x)", version);
|
||||
goto end;
|
||||
goto fault;
|
||||
}
|
||||
|
||||
while (1) {
|
||||
sid_t sid;
|
||||
r = rhizome_read_buffered(&read, &buff, sid.binary, sizeof sid.binary);
|
||||
if (r == 0) {
|
||||
status = MESHMS_STATUS_OK;
|
||||
goto end;
|
||||
}
|
||||
if (r != sizeof sid.binary)
|
||||
break;
|
||||
if (config.debug.meshms)
|
||||
@ -537,10 +581,11 @@ static int read_known_conversations(rhizome_manifest *m, const sid_t *their_sid,
|
||||
ofs += unpacked;
|
||||
read.offset += ofs - bytes;
|
||||
}
|
||||
ret = 0;
|
||||
fault:
|
||||
status = MESHMS_STATUS_PROTOCOL_FAULT;
|
||||
end:
|
||||
rhizome_read_close(&read);
|
||||
return ret;
|
||||
return status;
|
||||
}
|
||||
|
||||
static ssize_t write_conversation(struct rhizome_write *write, struct meshms_conversations *conv)
|
||||
@ -585,13 +630,13 @@ static ssize_t write_conversation(struct rhizome_write *write, struct meshms_con
|
||||
return len;
|
||||
}
|
||||
|
||||
static int write_known_conversations(rhizome_manifest *m, struct meshms_conversations *conv)
|
||||
static enum meshms_status write_known_conversations(rhizome_manifest *m, struct meshms_conversations *conv)
|
||||
{
|
||||
rhizome_manifest *mout=NULL;
|
||||
|
||||
struct rhizome_write write;
|
||||
bzero(&write, sizeof(write));
|
||||
int ret=-1;
|
||||
enum meshms_status status = MESHMS_STATUS_ERROR;
|
||||
|
||||
// TODO rebalance tree...
|
||||
|
||||
@ -617,66 +662,65 @@ static int write_known_conversations(rhizome_manifest *m, struct meshms_conversa
|
||||
goto end;
|
||||
rhizome_manifest_set_filehash(m, &write.id);
|
||||
}
|
||||
enum rhizome_bundle_status status = rhizome_manifest_finalise(m, &mout, 1);
|
||||
switch (status) {
|
||||
enum rhizome_bundle_status bstatus = rhizome_manifest_finalise(m, &mout, 1);
|
||||
switch (bstatus) {
|
||||
case RHIZOME_BUNDLE_STATUS_ERROR:
|
||||
// error is already logged
|
||||
break;
|
||||
case RHIZOME_BUNDLE_STATUS_NEW:
|
||||
ret = 0;
|
||||
status = MESHMS_STATUS_OK;
|
||||
break;
|
||||
case RHIZOME_BUNDLE_STATUS_SAME:
|
||||
case RHIZOME_BUNDLE_STATUS_DUPLICATE:
|
||||
case RHIZOME_BUNDLE_STATUS_OLD:
|
||||
WHYF("MeshMS conversation manifest (version=%"PRIu64") gazumped by Rhizome store (version=%"PRIu64")",
|
||||
m->version, mout->version);
|
||||
status = MESHMS_STATUS_PROTOCOL_FAULT;
|
||||
break;
|
||||
case RHIZOME_BUNDLE_STATUS_INCONSISTENT:
|
||||
WHY("MeshMS conversation manifest not consistent with payload");
|
||||
status = MESHMS_STATUS_PROTOCOL_FAULT;
|
||||
break;
|
||||
case RHIZOME_BUNDLE_STATUS_FAKE:
|
||||
WHY("MeshMS conversation manifest is not signed");
|
||||
status = MESHMS_STATUS_PROTOCOL_FAULT;
|
||||
break;
|
||||
case RHIZOME_BUNDLE_STATUS_INVALID:
|
||||
WHY("MeshMS conversation manifest is invalid");
|
||||
status = MESHMS_STATUS_PROTOCOL_FAULT;
|
||||
break;
|
||||
default:
|
||||
FATALF("status=%d", status);
|
||||
FATALF("bstatus=%d", bstatus);
|
||||
}
|
||||
end:
|
||||
if (ret)
|
||||
if (meshms_failed(status))
|
||||
rhizome_fail_write(&write);
|
||||
if (mout && m!=mout)
|
||||
rhizome_manifest_free(mout);
|
||||
return ret;
|
||||
return status;
|
||||
}
|
||||
|
||||
// read information about existing conversations from a rhizome payload
|
||||
int meshms_conversations_list(const sid_t *my_sid, const sid_t *their_sid, struct meshms_conversations **conv)
|
||||
enum meshms_status meshms_conversations_list(const sid_t *my_sid, const sid_t *their_sid, struct meshms_conversations **conv)
|
||||
{
|
||||
int ret=-1;
|
||||
enum meshms_status status = MESHMS_STATUS_ERROR;
|
||||
rhizome_manifest *m = rhizome_new_manifest();
|
||||
if (!m)
|
||||
goto end;
|
||||
if (get_my_conversation_bundle(my_sid, m))
|
||||
if (meshms_failed(status = get_my_conversation_bundle(my_sid, m)))
|
||||
goto end;
|
||||
|
||||
// read conversations payload
|
||||
if (read_known_conversations(m, their_sid, conv))
|
||||
if (meshms_failed(status = read_known_conversations(m, their_sid, conv)))
|
||||
goto end;
|
||||
|
||||
if (get_database_conversations(my_sid, their_sid, conv))
|
||||
if (meshms_failed(status = get_database_conversations(my_sid, their_sid, conv)))
|
||||
goto end;
|
||||
|
||||
if (update_conversations(my_sid, *conv) && !their_sid){
|
||||
if (write_known_conversations(m, *conv))
|
||||
goto end;
|
||||
}
|
||||
ret=0;
|
||||
|
||||
if ((status = update_conversations(my_sid, *conv)) == MESHMS_STATUS_UPDATED && their_sid == NULL)
|
||||
status = write_known_conversations(m, *conv);
|
||||
end:
|
||||
rhizome_manifest_free(m);
|
||||
return ret;
|
||||
if (config.debug.meshms)
|
||||
DEBUGF("status=%d", status);
|
||||
return status;
|
||||
}
|
||||
|
||||
/* Start traversing the given conversation binary tree in infix order.
|
||||
@ -715,11 +759,13 @@ void meshms_conversation_iterator_advance(struct meshms_conversation_iterator *i
|
||||
}
|
||||
}
|
||||
|
||||
int meshms_message_iterator_open(struct meshms_message_iterator *iter, const sid_t *me, const sid_t *them)
|
||||
enum meshms_status meshms_message_iterator_open(struct meshms_message_iterator *iter, const sid_t *me, const sid_t *them)
|
||||
{
|
||||
enum meshms_status status;
|
||||
bzero(iter, sizeof *iter);
|
||||
if ((iter->_conv = find_or_create_conv(me, them)) == NULL)
|
||||
if (meshms_failed(status = find_or_create_conv(me, them, &iter->_conv)))
|
||||
goto fail;
|
||||
assert(iter->_conv != NULL);
|
||||
iter->_my_sid = *me;
|
||||
iter->my_sid = &iter->_my_sid;
|
||||
iter->their_sid = &iter->_conv->them;
|
||||
@ -729,17 +775,18 @@ int meshms_message_iterator_open(struct meshms_message_iterator *iter, const sid
|
||||
// If I have never sent a message (or acked any of theirs), there are no messages in the thread.
|
||||
if (iter->_conv->found_my_ply) {
|
||||
if ((iter->_my_manifest = rhizome_new_manifest()) == NULL)
|
||||
goto fail;
|
||||
if (ply_read_open(&iter->_my_reader, &iter->_conv->my_ply.bundle_id, iter->_my_manifest))
|
||||
goto error;
|
||||
if (meshms_failed(status = ply_read_open(&iter->_my_reader, &iter->_conv->my_ply.bundle_id, iter->_my_manifest)))
|
||||
goto fail;
|
||||
if (iter->_conv->found_their_ply) {
|
||||
if ((iter->_their_manifest = rhizome_new_manifest()) == NULL)
|
||||
goto fail;
|
||||
if (ply_read_open(&iter->_their_reader, &iter->_conv->their_ply.bundle_id, iter->_their_manifest))
|
||||
goto error;
|
||||
if (meshms_failed(status = ply_read_open(&iter->_their_reader, &iter->_conv->their_ply.bundle_id, iter->_their_manifest)))
|
||||
goto fail;
|
||||
// Find their latest ACK so we know which of my messages have been delivered.
|
||||
int r = ply_find_prev(&iter->_their_reader, MESHMS_BLOCK_TYPE_ACK);
|
||||
if (r == 0) {
|
||||
if (meshms_failed(status = ply_find_prev(&iter->_their_reader, MESHMS_BLOCK_TYPE_ACK)))
|
||||
goto fail;
|
||||
if (status == MESHMS_STATUS_UPDATED) {
|
||||
if (unpack_uint(iter->_their_reader.record, iter->_their_reader.record_length, &iter->latest_ack_my_offset) == -1)
|
||||
iter->latest_ack_my_offset = 0;
|
||||
else
|
||||
@ -755,10 +802,12 @@ int meshms_message_iterator_open(struct meshms_message_iterator *iter, const sid
|
||||
DEBUGF("Did not find sender's ply; no messages in thread");
|
||||
}
|
||||
iter->_in_ack = 0;
|
||||
return 0;
|
||||
return MESHMS_STATUS_OK;
|
||||
error:
|
||||
status = MESHMS_STATUS_ERROR;
|
||||
fail:
|
||||
meshms_message_iterator_close(iter);
|
||||
return -1;
|
||||
return status;
|
||||
}
|
||||
|
||||
int meshms_message_iterator_is_open(const struct meshms_message_iterator *iter)
|
||||
@ -782,7 +831,7 @@ void meshms_message_iterator_close(struct meshms_message_iterator *iter)
|
||||
iter->_conv = NULL;
|
||||
}
|
||||
|
||||
int meshms_message_iterator_prev(struct meshms_message_iterator *iter)
|
||||
enum meshms_status meshms_message_iterator_prev(struct meshms_message_iterator *iter)
|
||||
{
|
||||
assert(iter->_conv != NULL);
|
||||
if (iter->_conv->found_my_ply) {
|
||||
@ -790,15 +839,15 @@ int meshms_message_iterator_prev(struct meshms_message_iterator *iter)
|
||||
if (iter->_conv->found_their_ply)
|
||||
assert(iter->_their_manifest != NULL);
|
||||
}
|
||||
int ret = 0;
|
||||
while (ret == 0) {
|
||||
enum meshms_status status = MESHMS_STATUS_UPDATED;
|
||||
while (status == MESHMS_STATUS_UPDATED) {
|
||||
if (iter->_in_ack) {
|
||||
if (config.debug.meshms)
|
||||
DEBUGF("Reading other log from %"PRId64", to %"PRId64, iter->_their_reader.read.offset, iter->_end_range);
|
||||
if ((ret = ply_read_prev(&iter->_their_reader)) == -1)
|
||||
if (meshms_failed(status = ply_read_prev(&iter->_their_reader)))
|
||||
break;
|
||||
iter->which_ply = THEIR_PLY;
|
||||
if (ret == 0 && iter->_their_reader.read.offset >= iter->_end_range) {
|
||||
if (status == MESHMS_STATUS_UPDATED && iter->_their_reader.read.offset >= iter->_end_range) {
|
||||
switch (iter->_their_reader.type) {
|
||||
case MESHMS_BLOCK_TYPE_ACK:
|
||||
iter->type = ACK_RECEIVED;
|
||||
@ -807,7 +856,7 @@ int meshms_message_iterator_prev(struct meshms_message_iterator *iter)
|
||||
iter->text_length = 0;
|
||||
if (unpack_uint(iter->_their_reader.record, iter->_their_reader.record_length, &iter->ack_offset) == -1)
|
||||
iter->ack_offset = 0;
|
||||
return 0;
|
||||
return status;
|
||||
case MESHMS_BLOCK_TYPE_MESSAGE:
|
||||
iter->type = MESSAGE_RECEIVED;
|
||||
iter->offset = iter->_their_reader.record_end_offset;
|
||||
@ -817,17 +866,17 @@ int meshms_message_iterator_prev(struct meshms_message_iterator *iter)
|
||||
&& iter->_their_reader.record[iter->_their_reader.record_length - 1] == '\0'
|
||||
) {
|
||||
iter->read = iter->_their_reader.record_end_offset <= iter->_conv->read_offset;
|
||||
return 0;
|
||||
return status;
|
||||
}
|
||||
WARN("Malformed MeshMS2 ply journal, missing NUL terminator");
|
||||
break;
|
||||
return MESHMS_STATUS_PROTOCOL_FAULT;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
iter->_in_ack = 0;
|
||||
ret = 0;
|
||||
status = MESHMS_STATUS_UPDATED;
|
||||
}
|
||||
else if ((ret = ply_read_prev(&iter->_my_reader)) == 0) {
|
||||
else if ((status = ply_read_prev(&iter->_my_reader)) == MESHMS_STATUS_UPDATED) {
|
||||
if (config.debug.meshms)
|
||||
DEBUGF("Offset %"PRId64", type %d, read_offset %"PRId64, iter->_my_reader.read.offset, iter->_my_reader.type, iter->read_offset);
|
||||
iter->which_ply = MY_PLY;
|
||||
@ -836,8 +885,10 @@ int meshms_message_iterator_prev(struct meshms_message_iterator *iter)
|
||||
// Read the received messages up to the ack'ed offset
|
||||
if (iter->_conv->found_their_ply) {
|
||||
int ofs = unpack_uint(iter->_my_reader.record, iter->_my_reader.record_length, (uint64_t*)&iter->_their_reader.read.offset);
|
||||
if (ofs == -1)
|
||||
return WHYF("Malformed ACK");
|
||||
if (ofs == -1) {
|
||||
WHYF("Malformed ACK");
|
||||
return MESHMS_STATUS_PROTOCOL_FAULT;
|
||||
}
|
||||
uint64_t end_range;
|
||||
int x = unpack_uint(iter->_my_reader.record + ofs, iter->_my_reader.record_length - ofs, &end_range);
|
||||
if (x == -1)
|
||||
@ -857,32 +908,36 @@ int meshms_message_iterator_prev(struct meshms_message_iterator *iter)
|
||||
iter->text = (const char *)iter->_my_reader.record;
|
||||
iter->text_length = iter->_my_reader.record_length;
|
||||
iter->delivered = iter->latest_ack_my_offset && iter->_my_reader.record_end_offset <= iter->latest_ack_my_offset;
|
||||
return 0;
|
||||
return status;
|
||||
}
|
||||
}
|
||||
}
|
||||
return ret;
|
||||
return status;
|
||||
}
|
||||
|
||||
int meshms_send_message(const sid_t *sender, const sid_t *recipient, const char *message, size_t message_len)
|
||||
enum meshms_status meshms_send_message(const sid_t *sender, const sid_t *recipient, const char *message, size_t message_len)
|
||||
{
|
||||
assert(message_len != 0);
|
||||
if (message_len > MESHMS_MESSAGE_MAX_LEN)
|
||||
return WHY("message too long");
|
||||
struct meshms_conversations *conv = find_or_create_conv(sender, recipient);
|
||||
if (!conv)
|
||||
return -1;
|
||||
// construct a message payload
|
||||
// TODO, new format here.
|
||||
unsigned char buffer[message_len + 4];
|
||||
strncpy((char*)buffer, message, message_len);
|
||||
// ensure message is NUL terminated
|
||||
if (message[message_len - 1] != '\0')
|
||||
buffer[message_len++] = '\0';
|
||||
message_len += append_footer(buffer + message_len, MESHMS_BLOCK_TYPE_MESSAGE, message_len);
|
||||
int ret = append_meshms_buffer(sender, conv, buffer, message_len);
|
||||
if (message_len > MESHMS_MESSAGE_MAX_LEN) {
|
||||
WHY("message too long");
|
||||
return MESHMS_STATUS_ERROR;
|
||||
}
|
||||
struct meshms_conversations *conv = NULL;
|
||||
enum meshms_status status;
|
||||
if (!meshms_failed(status = find_or_create_conv(sender, recipient, &conv))) {
|
||||
assert(conv != NULL);
|
||||
// construct a message payload
|
||||
// TODO, new format here.
|
||||
unsigned char buffer[message_len + 4];
|
||||
strncpy((char*)buffer, message, message_len);
|
||||
// ensure message is NUL terminated
|
||||
if (message[message_len - 1] != '\0')
|
||||
buffer[message_len++] = '\0';
|
||||
message_len += append_footer(buffer + message_len, MESHMS_BLOCK_TYPE_MESSAGE, message_len);
|
||||
status = append_meshms_buffer(sender, conv, buffer, message_len);
|
||||
}
|
||||
meshms_free_conversations(conv);
|
||||
return ret;
|
||||
return status;
|
||||
}
|
||||
|
||||
// output the list of existing conversations for a given local identity
|
||||
@ -910,10 +965,11 @@ int app_meshms_conversations(const struct cli_parsed *parsed, struct cli_context
|
||||
}
|
||||
|
||||
struct meshms_conversations *conv=NULL;
|
||||
if (meshms_conversations_list(&sid, NULL, &conv)){
|
||||
enum meshms_status status;
|
||||
if (meshms_failed(status = meshms_conversations_list(&sid, NULL, &conv))) {
|
||||
keyring_free(keyring);
|
||||
return -1;
|
||||
}
|
||||
return status;
|
||||
}
|
||||
const char *names[]={
|
||||
"_id","recipient","read", "last_message", "read_offset"
|
||||
};
|
||||
@ -965,9 +1021,9 @@ int app_meshms_send_message(const struct cli_parsed *parsed, struct cli_context
|
||||
if (str_to_sid_t(&their_sid, their_sidhex) == -1)
|
||||
return WHY("invalid recipient SID");
|
||||
// include terminating NUL
|
||||
int ret = meshms_send_message(&my_sid, &their_sid, message, strlen(message) + 1);
|
||||
enum meshms_status status = meshms_send_message(&my_sid, &their_sid, message, strlen(message) + 1);
|
||||
keyring_free(keyring);
|
||||
return ret;
|
||||
return meshms_failed(status) ? status : 0;
|
||||
}
|
||||
|
||||
int app_meshms_list_messages(const struct cli_parsed *parsed, struct cli_context *context)
|
||||
@ -994,9 +1050,10 @@ int app_meshms_list_messages(const struct cli_parsed *parsed, struct cli_context
|
||||
return WHY("invalid recipient SID");
|
||||
}
|
||||
struct meshms_message_iterator iter;
|
||||
if (meshms_message_iterator_open(&iter, &my_sid, &their_sid) == -1) {
|
||||
enum meshms_status status;
|
||||
if (meshms_failed(status = meshms_message_iterator_open(&iter, &my_sid, &their_sid))) {
|
||||
keyring_free(keyring);
|
||||
return -1;
|
||||
return status;
|
||||
}
|
||||
const char *names[]={
|
||||
"_id","offset","type","message"
|
||||
@ -1005,8 +1062,7 @@ int app_meshms_list_messages(const struct cli_parsed *parsed, struct cli_context
|
||||
bool_t marked_delivered = 0;
|
||||
bool_t marked_read = 0;
|
||||
int id = 0;
|
||||
int ret;
|
||||
while ((ret = meshms_message_iterator_prev(&iter)) == 0) {
|
||||
while ((status = meshms_message_iterator_prev(&iter)) == MESHMS_STATUS_UPDATED) {
|
||||
switch (iter.type) {
|
||||
case MESSAGE_SENT:
|
||||
if (iter.delivered && !marked_delivered){
|
||||
@ -1040,18 +1096,17 @@ int app_meshms_list_messages(const struct cli_parsed *parsed, struct cli_context
|
||||
break;
|
||||
}
|
||||
}
|
||||
if (ret != -1) {
|
||||
if (!meshms_failed(status))
|
||||
cli_row_count(context, id);
|
||||
ret = 0;
|
||||
}
|
||||
meshms_message_iterator_close(&iter);
|
||||
keyring_free(keyring);
|
||||
return ret;
|
||||
return status;
|
||||
}
|
||||
|
||||
static int mark_read(struct meshms_conversations *conv, const sid_t *their_sid, const char *offset_str)
|
||||
// Returns the number of read markers moved.
|
||||
static unsigned mark_read(struct meshms_conversations *conv, const sid_t *their_sid, const char *offset_str)
|
||||
{
|
||||
int ret=0;
|
||||
unsigned ret=0;
|
||||
if (conv){
|
||||
int cmp = their_sid ? cmp_sid_t(&conv->them, their_sid) : 0;
|
||||
if (!their_sid || cmp<0){
|
||||
@ -1089,7 +1144,7 @@ int app_meshms_mark_read(const struct cli_parsed *parsed, struct cli_context *UN
|
||||
|| cli_arg(parsed, "recipient_sid", &their_sidhex, str_is_subscriber_id, NULL) == -1
|
||||
|| cli_arg(parsed, "offset", &offset_str, NULL, NULL)==-1)
|
||||
return -1;
|
||||
|
||||
|
||||
if (create_serval_instance_dir() == -1)
|
||||
return -1;
|
||||
if (!(keyring = keyring_open_instance_cli(parsed)))
|
||||
@ -1097,44 +1152,39 @@ int app_meshms_mark_read(const struct cli_parsed *parsed, struct cli_context *UN
|
||||
if (rhizome_opendb() == -1){
|
||||
keyring_free(keyring);
|
||||
return -1;
|
||||
}
|
||||
}
|
||||
sid_t my_sid, their_sid;
|
||||
fromhex(my_sid.binary, my_sidhex, sizeof(my_sid.binary));
|
||||
if (their_sidhex)
|
||||
fromhex(their_sid.binary, their_sidhex, sizeof(their_sid.binary));
|
||||
|
||||
int ret=-1;
|
||||
struct meshms_conversations *conv=NULL;
|
||||
|
||||
enum meshms_status status = MESHMS_STATUS_ERROR;
|
||||
struct meshms_conversations *conv = NULL;
|
||||
rhizome_manifest *m = rhizome_new_manifest();
|
||||
if (!m)
|
||||
goto end;
|
||||
if (get_my_conversation_bundle(&my_sid, m))
|
||||
if (meshms_failed(status = get_my_conversation_bundle(&my_sid, m)))
|
||||
goto end;
|
||||
|
||||
// read all conversations, so we can write them again
|
||||
if (read_known_conversations(m, NULL, &conv))
|
||||
if (meshms_failed(status = read_known_conversations(m, NULL, &conv)))
|
||||
goto end;
|
||||
|
||||
// read the full list of conversations from the database too
|
||||
if (get_database_conversations(&my_sid, NULL, &conv))
|
||||
if (meshms_failed(status = get_database_conversations(&my_sid, NULL, &conv)))
|
||||
goto end;
|
||||
|
||||
// check if any incoming conversations need to be acked or have new messages and update the read offset
|
||||
int changed = update_conversations(&my_sid, conv);
|
||||
int changed = 0;
|
||||
if (meshms_failed(status = update_conversations(&my_sid, conv)))
|
||||
goto end;
|
||||
if (status == MESHMS_STATUS_UPDATED)
|
||||
changed = 1;
|
||||
if (mark_read(conv, their_sidhex?&their_sid:NULL, offset_str))
|
||||
changed =1;
|
||||
if (changed){
|
||||
// save the conversation list
|
||||
if (write_known_conversations(m, conv))
|
||||
goto end;
|
||||
}
|
||||
|
||||
ret=0;
|
||||
|
||||
if (changed)
|
||||
status = write_known_conversations(m, conv);
|
||||
end:
|
||||
if (m)
|
||||
rhizome_manifest_free(m);
|
||||
meshms_free_conversations(conv);
|
||||
keyring_free(keyring);
|
||||
return ret;
|
||||
return status;
|
||||
}
|
||||
|
||||
44
meshms.h
44
meshms.h
@ -20,11 +20,34 @@ Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
|
||||
#ifndef __SERVAL_DNA__MESHMS_H
|
||||
#define __SERVAL_DNA__MESHMS_H
|
||||
|
||||
#ifndef __MESHMS_INLINE
|
||||
# if __GNUC__ && !__GNUC_STDC_INLINE__
|
||||
# define __MESHMS_INLINE extern inline
|
||||
# else
|
||||
# define __MESHMS_INLINE inline
|
||||
# endif
|
||||
#endif
|
||||
|
||||
#include "serval.h"
|
||||
#include "rhizome.h"
|
||||
|
||||
#define MESHMS_MESSAGE_MAX_LEN 4095
|
||||
|
||||
/* The result of a MeshMS operation. Negative indicates failure, zero or
|
||||
* positive success.
|
||||
*/
|
||||
enum meshms_status {
|
||||
MESHMS_STATUS_ERROR = -1, // unexpected error (underlying failure)
|
||||
MESHMS_STATUS_OK = 0, // operation succeeded, no bundle changed
|
||||
MESHMS_STATUS_UPDATED = 1, // operation succeeded, bundle updated
|
||||
MESHMS_STATUS_SID_LOCKED = 2, // cannot decode or send messages for that SID
|
||||
MESHMS_STATUS_PROTOCOL_FAULT = 4, // missing or faulty ply bundle
|
||||
};
|
||||
|
||||
__MESHMS_INLINE int meshms_failed(enum meshms_status status) {
|
||||
return status != MESHMS_STATUS_OK && status != MESHMS_STATUS_UPDATED;
|
||||
}
|
||||
|
||||
// the manifest details for one half of a conversation
|
||||
struct meshms_ply {
|
||||
rhizome_bid_t bundle_id;
|
||||
@ -76,7 +99,7 @@ struct meshms_ply_read {
|
||||
/* Fetch the list of all MeshMS conversations into a binary tree whose nodes
|
||||
* are all allocated by malloc(3).
|
||||
*/
|
||||
int meshms_conversations_list(const sid_t *my_sid, const sid_t *their_sid, struct meshms_conversations **conv);
|
||||
enum meshms_status meshms_conversations_list(const sid_t *my_sid, const sid_t *their_sid, struct meshms_conversations **conv);
|
||||
void meshms_free_conversations(struct meshms_conversations *conv);
|
||||
|
||||
/* For iterating over a binary tree of all MeshMS conversations, as created by
|
||||
@ -96,17 +119,20 @@ void meshms_conversation_iterator_start(struct meshms_conversation_iterator *, s
|
||||
void meshms_conversation_iterator_advance(struct meshms_conversation_iterator *);
|
||||
|
||||
/* For iterating through the messages in a single MeshMS conversation; both
|
||||
* plys threaded (interleaved) in the order as seen by the sender.
|
||||
* plys threaded (interleaved) in the order as seen by the sender. The
|
||||
* meshms_message_iterator_prev() function returns MESHMS_STATUS_UPDATED if it
|
||||
* advances the iterator to a message, or MESHMS_STATUS_OK if there are no more
|
||||
* messages. Any other return value indicates failure.
|
||||
*
|
||||
* struct meshms_message_iterator it;
|
||||
* if (meshms_message_iterator_open(&it, &sender_sid, &recip_sid) == -1)
|
||||
* enum meshms_status status;
|
||||
* if (meshms_failed(status = meshms_message_iterator_open(&it, &sender_sid, &recip_sid)))
|
||||
* return -1;
|
||||
* int ret;
|
||||
* while ((ret = meshms_message_iterator_prev(&it)) == 0) {
|
||||
* while ((status = meshms_message_iterator_prev(&it)) == MESHMS_STATUS_UPDATED) {
|
||||
* ...
|
||||
* }
|
||||
* meshms_message_iterator_close(&it);
|
||||
* if (ret == -1)
|
||||
* if (meshms_failed(status))
|
||||
* return -1;
|
||||
* ...
|
||||
*/
|
||||
@ -143,16 +169,16 @@ struct meshms_message_iterator {
|
||||
uint64_t _end_range;
|
||||
bool_t _in_ack;
|
||||
};
|
||||
int meshms_message_iterator_open(struct meshms_message_iterator *, const sid_t *me, const sid_t *them);
|
||||
enum meshms_status meshms_message_iterator_open(struct meshms_message_iterator *, const sid_t *me, const sid_t *them);
|
||||
int meshms_message_iterator_is_open(const struct meshms_message_iterator *);
|
||||
void meshms_message_iterator_close(struct meshms_message_iterator *);
|
||||
int meshms_message_iterator_prev(struct meshms_message_iterator *);
|
||||
enum meshms_status meshms_message_iterator_prev(struct meshms_message_iterator *);
|
||||
|
||||
/* Append a message ('message_len' bytes of UTF8 at 'message') to the sender's
|
||||
* ply in the conversation between 'sender' and 'recipient'. If no
|
||||
* conversation (ply bundle) exists, then create it. Returns 0 on success, -1
|
||||
* on error (already logged).
|
||||
*/
|
||||
int meshms_send_message(const sid_t *sender, const sid_t *recipient, const char *message, size_t message_len);
|
||||
enum meshms_status meshms_send_message(const sid_t *sender, const sid_t *recipient, const char *message, size_t message_len);
|
||||
|
||||
#endif // __SERVAL_DNA__MESHMS_H
|
||||
|
||||
@ -208,11 +208,13 @@ static HTTP_CONTENT_GENERATOR restful_meshms_messagelist_json_content;
|
||||
|
||||
static int reopen_meshms_message_iterator(httpd_request *r)
|
||||
{
|
||||
enum meshms_status status;
|
||||
if (!meshms_message_iterator_is_open(&r->u.msglist.iter)) {
|
||||
if ( meshms_message_iterator_open(&r->u.msglist.iter, &r->sid1, &r->sid2) == -1
|
||||
|| (r->u.msglist.finished = meshms_message_iterator_prev(&r->u.msglist.iter)) == -1
|
||||
if ( meshms_failed(status = meshms_message_iterator_open(&r->u.msglist.iter, &r->sid1, &r->sid2))
|
||||
|| meshms_failed(status = meshms_message_iterator_prev(&r->u.msglist.iter))
|
||||
)
|
||||
return -1;
|
||||
r->u.msglist.finished = status != MESHMS_STATUS_UPDATED;
|
||||
if (!r->u.msglist.finished) {
|
||||
r->u.msglist.latest_which_ply = r->u.msglist.iter.which_ply;
|
||||
r->u.msglist.latest_offset = r->u.msglist.iter.offset;
|
||||
@ -399,8 +401,10 @@ static int restful_meshms_messagelist_json_content_chunk(struct http_request *hr
|
||||
}
|
||||
if (!strbuf_overrun(b)) {
|
||||
++r->u.msglist.rowcount;
|
||||
if ((r->u.msglist.finished = meshms_message_iterator_prev(&r->u.msglist.iter)) == -1)
|
||||
enum meshms_status status;
|
||||
if (meshms_failed(status = meshms_message_iterator_prev(&r->u.msglist.iter)))
|
||||
return -1;
|
||||
r->u.msglist.finished = status != MESHMS_STATUS_UPDATED;
|
||||
}
|
||||
return 1;
|
||||
}
|
||||
@ -498,9 +502,21 @@ static int restful_meshms_sendmessage_end(struct http_request *hr)
|
||||
return http_response_form_part(r, "Missing", PART_MESSAGE, NULL, 0);
|
||||
assert(r->u.sendmsg.message.length > 0);
|
||||
assert(r->u.sendmsg.message.length <= MESHMS_MESSAGE_MAX_LEN);
|
||||
int ret = meshms_send_message(&r->sid1, &r->sid2, r->u.sendmsg.message.buffer, r->u.sendmsg.message.length);
|
||||
if (ret == -1)
|
||||
return 500;
|
||||
http_request_simple_response(&r->http, 201, "Message sent");
|
||||
return 201;
|
||||
enum meshms_status status = meshms_send_message(&r->sid1, &r->sid2, r->u.sendmsg.message.buffer, r->u.sendmsg.message.length);
|
||||
switch (status) {
|
||||
case MESHMS_STATUS_ERROR:
|
||||
return 500;
|
||||
case MESHMS_STATUS_OK:
|
||||
case MESHMS_STATUS_UPDATED:
|
||||
http_request_simple_response(&r->http, 201, "Message sent");
|
||||
return 201;
|
||||
case MESHMS_STATUS_SID_LOCKED:
|
||||
http_request_simple_response(&r->http, 403, "Identity unknown");
|
||||
return 403;
|
||||
case MESHMS_STATUS_PROTOCOL_FAULT:
|
||||
http_request_simple_response(&r->http, 403, "Protocol fault");
|
||||
return 403;
|
||||
default:
|
||||
return 500;
|
||||
}
|
||||
}
|
||||
|
||||
17
tests/meshms
17
tests/meshms
@ -201,4 +201,21 @@ test_listConversations() {
|
||||
assertStdoutLineCount '==' 5
|
||||
}
|
||||
|
||||
doc_sendNoIdentity="Send message from unknown identity"
|
||||
setup_sendNoIdentity() {
|
||||
setup_servald
|
||||
set_instance +A
|
||||
create_identities 2
|
||||
setup_logging
|
||||
SIDX=0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF
|
||||
}
|
||||
test_sendNoIdentity() {
|
||||
executeOk_servald meshms send message $SIDA1 $SIDX "First message"
|
||||
execute $servald meshms send message $SIDX $SIDA1 "First reply"
|
||||
assertExitStatus --stderr == 2
|
||||
executeOk_servald meshms list messages $SIDA1 $SIDX
|
||||
execute $servald meshms list messages $SIDX $SIDA1
|
||||
assertExitStatus --stderr == 2
|
||||
}
|
||||
|
||||
runTests "$@"
|
||||
|
||||
@ -621,7 +621,7 @@ setup_RhizomeInsertLarge() {
|
||||
create_file file1 50m
|
||||
}
|
||||
test_RhizomeInsertLarge() {
|
||||
execute curl \
|
||||
execute --timeout=120 curl \
|
||||
--silent --show-error --write-out '%{http_code}' \
|
||||
--output file1.manifest \
|
||||
--dump-header http.header \
|
||||
@ -1220,4 +1220,69 @@ test_MeshmsSend() {
|
||||
assertStdoutGrep --matches=1 ':<:Hello back!$'
|
||||
}
|
||||
|
||||
doc_MeshmsSendMissingMessage="HTTP RESTful MeshMS send missing 'message' form part "
|
||||
setup_MeshmsSendMissingMessage() {
|
||||
IDENTITY_COUNT=2
|
||||
setup
|
||||
}
|
||||
test_MeshmsSendMissingMessage() {
|
||||
execute curl \
|
||||
--silent --show-error --write-out '%{http_code}' \
|
||||
--output http.body \
|
||||
--dump-header http.header \
|
||||
--basic --user harry:potter \
|
||||
--data '' \
|
||||
"http://$addr_localhost:$PORTA/restful/meshms/$SIDA2/$SIDA1/sendmessage"
|
||||
tfw_cat http.header http.body
|
||||
assertExitStatus == 0
|
||||
assertStdoutIs 403
|
||||
assertJq http.body 'contains({"http_status_code": 403})'
|
||||
assertJqGrep --ignore-case http.body '.http_status_message' 'missing.*message.*form.*part'
|
||||
executeOk_servald meshms list messages $SIDA1 $SIDA2
|
||||
assertStdoutLineCount '==' 2
|
||||
}
|
||||
|
||||
doc_MeshmsSendDuplicateMessage="HTTP RESTful MeshMS send missing 'message' form part "
|
||||
setup_MeshmsSendDuplicateMessage() {
|
||||
IDENTITY_COUNT=2
|
||||
setup
|
||||
}
|
||||
test_MeshmsSendDuplicateMessage() {
|
||||
execute curl \
|
||||
--silent --show-error --write-out '%{http_code}' \
|
||||
--output http.body \
|
||||
--dump-header http.header \
|
||||
--basic --user harry:potter \
|
||||
--form "message=Hello one" \
|
||||
--form "message=Hello two" \
|
||||
"http://$addr_localhost:$PORTA/restful/meshms/$SIDA2/$SIDA1/sendmessage"
|
||||
tfw_cat http.header http.body
|
||||
assertExitStatus == 0
|
||||
assertStdoutIs 403
|
||||
assertJq http.body 'contains({"http_status_code": 403})'
|
||||
assertJqGrep --ignore-case http.body '.http_status_message' 'duplicate.*message.*form.*part'
|
||||
executeOk_servald meshms list messages $SIDA1 $SIDA2
|
||||
assertStdoutLineCount '==' 2
|
||||
}
|
||||
|
||||
doc_MeshmsSendNoIdentity="HTTP RESTful MeshMS send from unknown identity"
|
||||
setup_MeshmsSendNoIdentity() {
|
||||
setup
|
||||
SIDX=0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF
|
||||
}
|
||||
test_MeshmsSendNoIdentity() {
|
||||
execute curl \
|
||||
--silent --show-error --write-out '%{http_code}' \
|
||||
--output http.body \
|
||||
--dump-header http.header \
|
||||
--basic --user harry:potter \
|
||||
--form "message=Hello" \
|
||||
"http://$addr_localhost:$PORTA/restful/meshms/$SIDX/$SIDA/sendmessage"
|
||||
tfw_cat http.header http.body
|
||||
assertExitStatus == 0
|
||||
assertStdoutIs 403
|
||||
assertJq http.body 'contains({"http_status_code": 403})'
|
||||
assertJqGrep --ignore-case http.body '.http_status_message' 'identity.*unknown'
|
||||
}
|
||||
|
||||
runTests "$@"
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user