diff --git a/mongoose.c b/mongoose.c index 3dd88338..8326cdce 100644 --- a/mongoose.c +++ b/mongoose.c @@ -1793,14 +1793,14 @@ int mg_http_parse(const char *s, size_t len, struct mg_http_message *hm) { static void mg_http_vprintf_chunk(struct mg_connection *c, const char *fmt, va_list *ap) { size_t len = c->send.len; - mg_send(c, " \r\n", 10); + if (!mg_send(c, " \r\n", 10)) mg_error(c, "OOM"); mg_vxprintf(mg_pfn_iobuf, &c->send, fmt, ap); if (c->send.len >= len + 10) { mg_snprintf((char *) c->send.buf + len, 9, "%08lx", c->send.len - len - 10); c->send.buf[len + 8] = '\r'; if (c->send.len == len + 10) c->is_resp = 0; // Last chunk, reset marker } - mg_send(c, "\r\n", 2); + if (!mg_send(c, "\r\n", 2)) mg_error(c, "OOM"); } void mg_http_printf_chunk(struct mg_connection *c, const char *fmt, ...) { @@ -1812,8 +1812,7 @@ void mg_http_printf_chunk(struct mg_connection *c, const char *fmt, ...) { void mg_http_write_chunk(struct mg_connection *c, const char *buf, size_t len) { mg_printf(c, "%lx\r\n", (unsigned long) len); - mg_send(c, buf, len); - mg_send(c, "\r\n", 2); + if (!mg_send(c, buf, len) || !mg_send(c, "\r\n", 2)) mg_error(c, "OOM"); if (len == 0) c->is_resp = 0; } @@ -3353,8 +3352,8 @@ static const struct mg_mqtt_pmap s_prop_map[] = { {MQTT_PROP_SUBSCRIPTION_IDENTIFIER_AVAILABLE, MQTT_PROP_TYPE_BYTE}, {MQTT_PROP_SHARED_SUBSCRIPTION_AVAILABLE, MQTT_PROP_TYPE_BYTE}}; -void mg_mqtt_send_header(struct mg_connection *c, uint8_t cmd, uint8_t flags, - uint32_t len) { +static bool mqtt_send_header(struct mg_connection *c, uint8_t cmd, + uint8_t flags, uint32_t len) { uint8_t buf[1 + sizeof(len)], *vlen = &buf[1]; buf[0] = (uint8_t) ((cmd << 4) | flags); do { @@ -3363,15 +3362,20 @@ void mg_mqtt_send_header(struct mg_connection *c, uint8_t cmd, uint8_t flags, if (len > 0) *vlen |= 0x80; vlen++; } while (len > 0 && vlen < &buf[sizeof(buf)]); - mg_send(c, buf, (size_t) (vlen - buf)); + return mg_send(c, buf, (size_t) (vlen - buf)); } -static void mg_send_u16(struct mg_connection *c, uint16_t value) { - mg_send(c, &value, sizeof(value)); +void mg_mqtt_send_header(struct mg_connection *c, uint8_t cmd, uint8_t flags, + uint32_t len) { + if (!mqtt_send_header(c, cmd, flags, len)) mg_error(c, "OOM"); } -static void mg_send_u32(struct mg_connection *c, uint32_t value) { - mg_send(c, &value, sizeof(value)); +static bool mg_send_u16(struct mg_connection *c, uint16_t value) { + return mg_send(c, &value, sizeof(value)); +} + +static bool mg_send_u32(struct mg_connection *c, uint32_t value) { + return mg_send(c, &value, sizeof(value)); } static uint8_t varint_size(size_t length) { @@ -3464,46 +3468,50 @@ static size_t get_props_size(struct mg_mqtt_prop *props, size_t count) { return size; } -static void mg_send_mqtt_properties(struct mg_connection *c, +static bool mg_send_mqtt_properties(struct mg_connection *c, struct mg_mqtt_prop *props, size_t nprops) { size_t total_size = get_properties_length(props, nprops); uint8_t buf_v[4] = {0, 0, 0, 0}; uint8_t buf[4] = {0, 0, 0, 0}; size_t i, len = encode_varint(buf, total_size); - mg_send(c, buf, (size_t) len); + if (!mg_send(c, buf, (size_t) len)) return false; for (i = 0; i < nprops; i++) { - mg_send(c, &props[i].id, sizeof(props[i].id)); + if (!mg_send(c, &props[i].id, sizeof(props[i].id))) return false; switch (mqtt_prop_type_by_id(props[i].id)) { case MQTT_PROP_TYPE_STRING_PAIR: - mg_send_u16(c, mg_htons((uint16_t) props[i].key.len)); - mg_send(c, props[i].key.buf, props[i].key.len); - mg_send_u16(c, mg_htons((uint16_t) props[i].val.len)); - mg_send(c, props[i].val.buf, props[i].val.len); + if (!mg_send_u16(c, mg_htons((uint16_t) props[i].key.len)) || + !mg_send(c, props[i].key.buf, props[i].key.len) || + !mg_send_u16(c, mg_htons((uint16_t) props[i].val.len)) || + !mg_send(c, props[i].val.buf, props[i].val.len)) + return false; break; case MQTT_PROP_TYPE_BYTE: - mg_send(c, &props[i].iv, sizeof(uint8_t)); + if (!mg_send(c, &props[i].iv, sizeof(uint8_t))) return false; break; case MQTT_PROP_TYPE_SHORT: - mg_send_u16(c, mg_htons((uint16_t) props[i].iv)); + if (!mg_send_u16(c, mg_htons((uint16_t) props[i].iv))) return false; break; case MQTT_PROP_TYPE_INT: - mg_send_u32(c, mg_htonl((uint32_t) props[i].iv)); + if (!mg_send_u32(c, mg_htonl((uint32_t) props[i].iv))) return false; break; case MQTT_PROP_TYPE_STRING: - mg_send_u16(c, mg_htons((uint16_t) props[i].val.len)); - mg_send(c, props[i].val.buf, props[i].val.len); + if (!mg_send_u16(c, mg_htons((uint16_t) props[i].val.len)) || + !mg_send(c, props[i].val.buf, props[i].val.len)) + return false; break; case MQTT_PROP_TYPE_BINARY_DATA: - mg_send_u16(c, mg_htons((uint16_t) props[i].val.len)); - mg_send(c, props[i].val.buf, props[i].val.len); + if (!mg_send_u16(c, mg_htons((uint16_t) props[i].val.len)) || + !mg_send(c, props[i].val.buf, props[i].val.len)) + return false; break; case MQTT_PROP_TYPE_VARIABLE_INT: len = encode_varint(buf_v, props[i].iv); - mg_send(c, buf_v, (size_t) len); + if (!mg_send(c, buf_v, (size_t) len)) return false; break; } } + return true; } size_t mg_mqtt_next_prop(struct mg_mqtt_message *msg, struct mg_mqtt_prop *prop, @@ -3596,33 +3604,41 @@ void mg_mqtt_login(struct mg_connection *c, const struct mg_mqtt_opts *opts) { total_len += get_props_size(opts->will_props, opts->num_will_props); } - mg_mqtt_send_header(c, MQTT_CMD_CONNECT, 0, (uint32_t) total_len); - mg_send(c, hdr, sizeof(hdr)); // keepalive == 0 means "do not disconnect us!" - mg_send_u16(c, mg_htons((uint16_t) opts->keepalive)); + if (!mqtt_send_header(c, MQTT_CMD_CONNECT, 0, (uint32_t) total_len) || + !mg_send(c, hdr, sizeof(hdr)) || + !mg_send_u16(c, mg_htons((uint16_t) opts->keepalive))) + goto fail; - if (c->is_mqtt5) mg_send_mqtt_properties(c, opts->props, opts->num_props); + if (c->is_mqtt5 && !mg_send_mqtt_properties(c, opts->props, opts->num_props)) + goto fail; - mg_send_u16(c, mg_htons((uint16_t) cid.len)); - mg_send(c, cid.buf, cid.len); + if (!mg_send_u16(c, mg_htons((uint16_t) cid.len)) || + !mg_send(c, cid.buf, cid.len)) + goto fail; if (hdr[7] & MQTT_HAS_WILL) { - if (c->is_mqtt5) - mg_send_mqtt_properties(c, opts->will_props, opts->num_will_props); + if (c->is_mqtt5 && + !mg_send_mqtt_properties(c, opts->will_props, opts->num_will_props)) + goto fail; - mg_send_u16(c, mg_htons((uint16_t) opts->topic.len)); - mg_send(c, opts->topic.buf, opts->topic.len); - mg_send_u16(c, mg_htons((uint16_t) opts->message.len)); - mg_send(c, opts->message.buf, opts->message.len); - } - if (opts->user.len > 0) { - mg_send_u16(c, mg_htons((uint16_t) opts->user.len)); - mg_send(c, opts->user.buf, opts->user.len); - } - if (opts->pass.len > 0) { - mg_send_u16(c, mg_htons((uint16_t) opts->pass.len)); - mg_send(c, opts->pass.buf, opts->pass.len); + if (!mg_send_u16(c, mg_htons((uint16_t) opts->topic.len)) || + !mg_send(c, opts->topic.buf, opts->topic.len) || + !mg_send_u16(c, mg_htons((uint16_t) opts->message.len)) || + !mg_send(c, opts->message.buf, opts->message.len)) + goto fail; } + if (opts->user.len > 0 && + (!mg_send_u16(c, mg_htons((uint16_t) opts->user.len)) || + !mg_send(c, opts->user.buf, opts->user.len))) + goto fail; + if (opts->pass.len > 0 && + (!mg_send_u16(c, mg_htons((uint16_t) opts->pass.len)) || + !mg_send(c, opts->pass.buf, opts->pass.len))) + goto fail; + return; +fail: + mg_error(c, "OOM"); } uint16_t mg_mqtt_pub(struct mg_connection *c, const struct mg_mqtt_opts *opts) { @@ -3637,20 +3653,28 @@ uint16_t mg_mqtt_pub(struct mg_connection *c, const struct mg_mqtt_opts *opts) { if (c->is_mqtt5) len += get_props_size(opts->props, opts->num_props); if (opts->qos > 0 && id != 0) flags |= 1 << 3; - mg_mqtt_send_header(c, MQTT_CMD_PUBLISH, flags, (uint32_t) len); - mg_send_u16(c, mg_htons((uint16_t) opts->topic.len)); - mg_send(c, opts->topic.buf, opts->topic.len); + if (!mqtt_send_header(c, MQTT_CMD_PUBLISH, flags, (uint32_t) len) || + !mg_send_u16(c, mg_htons((uint16_t) opts->topic.len)) || + !mg_send(c, opts->topic.buf, opts->topic.len)) + goto fail; if (opts->qos > 0) { // need to send 'id' field if (id == 0) { // generate new one if not resending if (++c->mgr->mqtt_id == 0) ++c->mgr->mqtt_id; id = c->mgr->mqtt_id; } - mg_send_u16(c, mg_htons(id)); + if (!mg_send_u16(c, mg_htons(id))) goto fail; } - if (c->is_mqtt5) mg_send_mqtt_properties(c, opts->props, opts->num_props); + if (c->is_mqtt5 && !mg_send_mqtt_properties(c, opts->props, opts->num_props)) + goto fail; - if (opts->message.len > 0) mg_send(c, opts->message.buf, opts->message.len); + if (opts->message.len > 0 && + !mg_send(c, opts->message.buf, opts->message.len)) + goto fail; + return id; + +fail: + mg_error(c, "OOM"); return id; } @@ -3659,14 +3683,20 @@ void mg_mqtt_sub(struct mg_connection *c, const struct mg_mqtt_opts *opts) { size_t plen = c->is_mqtt5 ? get_props_size(opts->props, opts->num_props) : 0; size_t len = 2 + opts->topic.len + 2 + 1 + plen; - mg_mqtt_send_header(c, MQTT_CMD_SUBSCRIBE, 2, (uint32_t) len); + if (!mqtt_send_header(c, MQTT_CMD_SUBSCRIBE, 2, (uint32_t) len)) goto fail; if (++c->mgr->mqtt_id == 0) ++c->mgr->mqtt_id; - mg_send_u16(c, mg_htons(c->mgr->mqtt_id)); - if (c->is_mqtt5) mg_send_mqtt_properties(c, opts->props, opts->num_props); + if (!mg_send_u16(c, mg_htons(c->mgr->mqtt_id))) goto fail; - mg_send_u16(c, mg_htons((uint16_t) opts->topic.len)); - mg_send(c, opts->topic.buf, opts->topic.len); - mg_send(c, &qos_, sizeof(qos_)); + if (c->is_mqtt5 && !mg_send_mqtt_properties(c, opts->props, opts->num_props)) + goto fail; + + if (!mg_send_u16(c, mg_htons((uint16_t) opts->topic.len)) || + !mg_send(c, opts->topic.buf, opts->topic.len) || + !mg_send(c, &qos_, sizeof(qos_))) + goto fail; + return; +fail: + mg_error(c, "OOM"); } int mg_mqtt_parse(const uint8_t *buf, size_t len, uint8_t version, @@ -3773,15 +3803,16 @@ static void mqtt_cb(struct mg_connection *c, int ev, void *ev_data) { uint32_t remaining_len = sizeof(id); if (c->is_mqtt5) remaining_len += 2; // 3.4.2 - mg_mqtt_send_header( - c, - (uint8_t) (mm.qos == 2 ? MQTT_CMD_PUBREC : MQTT_CMD_PUBACK), - 0, remaining_len); - mg_send(c, &id, sizeof(id)); + if (!mqtt_send_header(c, + (uint8_t) (mm.qos == 2 ? MQTT_CMD_PUBREC + : MQTT_CMD_PUBACK), + 0, remaining_len) || + !mg_send(c, &id, sizeof(id))) + goto fail; if (c->is_mqtt5) { uint16_t zero = 0; - mg_send(c, &zero, sizeof(zero)); + if (!mg_send(c, &zero, sizeof(zero))) goto fail; } } mg_call(c, MG_EV_MQTT_MSG, &mm); // let the app handle qos stuff @@ -3790,15 +3821,18 @@ static void mqtt_cb(struct mg_connection *c, int ev, void *ev_data) { case MQTT_CMD_PUBREC: { // MQTT5: 3.5.2-1 TODO(): variable header rc uint16_t id = mg_ntohs(mm.id); uint32_t remaining_len = sizeof(id); // MQTT5 3.6.2-1 - mg_mqtt_send_header(c, MQTT_CMD_PUBREL, 2, remaining_len); - mg_send(c, &id, sizeof(id)); // MQTT5 3.6.1-1, flags = 2 + if (!mqtt_send_header(c, MQTT_CMD_PUBREL, 2, + remaining_len) // MQTT5 3.6.1-1, flags = 2 + || !mg_send(c, &id, sizeof(id))) + goto fail; break; } case MQTT_CMD_PUBREL: { // MQTT5: 3.6.2-1 TODO(): variable header rc uint16_t id = mg_ntohs(mm.id); uint32_t remaining_len = sizeof(id); // MQTT5 3.7.2-1 - mg_mqtt_send_header(c, MQTT_CMD_PUBCOMP, 0, remaining_len); - mg_send(c, &id, sizeof(id)); + if (!mqtt_send_header(c, MQTT_CMD_PUBCOMP, 0, remaining_len) || + !mg_send(c, &id, sizeof(id))) + goto fail; break; } } @@ -3810,6 +3844,9 @@ static void mqtt_cb(struct mg_connection *c, int ev, void *ev_data) { } } (void) ev_data; + return; +fail: + mg_error(c, "OOM"); } void mg_mqtt_ping(struct mg_connection *nc) { @@ -3824,19 +3861,24 @@ void mg_mqtt_disconnect(struct mg_connection *c, const struct mg_mqtt_opts *opts) { size_t len = 0; if (c->is_mqtt5) len = 1 + get_props_size(opts->props, opts->num_props); - mg_mqtt_send_header(c, MQTT_CMD_DISCONNECT, 0, (uint32_t) len); + if (!mqtt_send_header(c, MQTT_CMD_DISCONNECT, 0, (uint32_t) len)) goto fail; if (c->is_mqtt5) { uint8_t zero = 0; - mg_send(c, &zero, sizeof(zero)); // reason code - mg_send_mqtt_properties(c, opts->props, opts->num_props); + if (!mg_send(c, &zero, sizeof(zero)) // reason code + || !mg_send_mqtt_properties(c, opts->props, opts->num_props)) + goto fail; } + return; +fail: + mg_error(c, "OOM"); } struct mg_connection *mg_mqtt_connect(struct mg_mgr *mgr, const char *url, const struct mg_mqtt_opts *opts, mg_event_handler_t fn, void *fn_data) { - struct mg_connection *c = mg_connect_svc(mgr, url, fn, fn_data, mqtt_cb, NULL); + struct mg_connection *c = + mg_connect_svc(mgr, url, fn, fn_data, mqtt_cb, NULL); if (c != NULL) { struct mg_mqtt_opts empty; memset(&empty, 0, sizeof(empty)); @@ -6088,6 +6130,7 @@ bool mg_open_listener(struct mg_connection *c, const char *url) { static void write_conn(struct mg_connection *c) { long len = c->is_tls ? mg_tls_send(c, c->send.buf, c->send.len) : mg_io_send(c, c->send.buf, c->send.len); + // TODO(): mg_tls_send() may return 0 forever on steady OOM if (len == MG_IO_ERR) { mg_error(c, "tx err"); } else if (len > 0) { @@ -6100,14 +6143,8 @@ static void init_closure(struct mg_connection *c) { struct connstate *s = (struct connstate *) (c + 1); if (c->is_udp == false && c->is_listening == false && c->is_connecting == false) { // For TCP conns, -<<<<<<< HEAD tx_tcp(c->mgr->ifp, s->mac, &c->loc, &c->rem, TH_FIN | TH_ACK, mg_htonl(s->seq), mg_htonl(s->ack), NULL, 0); -======= - uint32_t rem_ip = c->rem.ip4; - tx_tcp(c->mgr->ifp, s->mac, rem_ip, TH_FIN | TH_ACK, c->loc.port, - c->rem.port, mg_htonl(s->seq), mg_htonl(s->ack), NULL, 0); ->>>>>>> 81b87991 (wip) settmout(c, MIP_TTYPE_FIN); } } @@ -6173,7 +6210,7 @@ bool mg_send(struct mg_connection *c, const void *buf, size_t len) { len = trim_len(c, len); // Trimming length if necessary res = tx_udp(ifp, s->mac, &c->loc, &c->rem, buf, len); } else { - res = mg_iobuf_add(&c->send, c->send.len, buf, len); + res = (bool) mg_iobuf_add(&c->send, c->send.len, buf, len); // res == 0 means an OOM condition (iobuf couldn't resize), yet this is so // far recoverable, let the caller decide } @@ -9530,7 +9567,7 @@ bool mg_send(struct mg_connection *c, const void *buf, size_t len) { iolog(c, (char *) buf, n, false); return n > 0; } else { - return mg_iobuf_add(&c->send, c->send.len, buf, len); + return (bool) mg_iobuf_add(&c->send, c->send.len, buf, len); // returning 0 means an OOM condition (iobuf couldn't resize), yet this is // so far recoverable, let the caller decide } @@ -9730,6 +9767,7 @@ static void write_conn(struct mg_connection *c) { char *buf = (char *) c->send.buf; size_t len = c->send.len; long n = c->is_tls ? mg_tls_send(c, buf, len) : mg_io_send(c, buf, len); + // TODO(): mg_tls_send() may return 0 forever on steady OOM MG_DEBUG(("%lu %ld snd %ld/%ld rcv %ld/%ld n=%ld err=%d", c->id, c->fd, (long) c->send.len, (long) c->send.size, (long) c->recv.len, (long) c->recv.size, n, MG_SOCK_ERR(n))); @@ -20932,7 +20970,7 @@ static void ws_handshake(struct mg_connection *c, const struct mg_str *wskey, mg_printf(c, "Sec-WebSocket-Protocol: %.*s\r\n", (int) wsproto->len, wsproto->buf); } - mg_send(c, "\r\n", 2); + if (!mg_send(c, "\r\n", 2)) mg_error(c, "OOM"); } static uint32_t be32(const uint8_t *p) { diff --git a/src/http.c b/src/http.c index d2080b6c..0f287e95 100644 --- a/src/http.c +++ b/src/http.c @@ -346,14 +346,14 @@ int mg_http_parse(const char *s, size_t len, struct mg_http_message *hm) { static void mg_http_vprintf_chunk(struct mg_connection *c, const char *fmt, va_list *ap) { size_t len = c->send.len; - mg_send(c, " \r\n", 10); + if (!mg_send(c, " \r\n", 10)) mg_error(c, "OOM"); mg_vxprintf(mg_pfn_iobuf, &c->send, fmt, ap); if (c->send.len >= len + 10) { mg_snprintf((char *) c->send.buf + len, 9, "%08lx", c->send.len - len - 10); c->send.buf[len + 8] = '\r'; if (c->send.len == len + 10) c->is_resp = 0; // Last chunk, reset marker } - mg_send(c, "\r\n", 2); + if (!mg_send(c, "\r\n", 2)) mg_error(c, "OOM"); } void mg_http_printf_chunk(struct mg_connection *c, const char *fmt, ...) { @@ -365,8 +365,7 @@ void mg_http_printf_chunk(struct mg_connection *c, const char *fmt, ...) { void mg_http_write_chunk(struct mg_connection *c, const char *buf, size_t len) { mg_printf(c, "%lx\r\n", (unsigned long) len); - mg_send(c, buf, len); - mg_send(c, "\r\n", 2); + if (!mg_send(c, buf, len) || !mg_send(c, "\r\n", 2)) mg_error(c, "OOM"); if (len == 0) c->is_resp = 0; } diff --git a/src/mqtt.c b/src/mqtt.c index 6d96f2c0..9e7f1481 100644 --- a/src/mqtt.c +++ b/src/mqtt.c @@ -46,8 +46,8 @@ static const struct mg_mqtt_pmap s_prop_map[] = { {MQTT_PROP_SUBSCRIPTION_IDENTIFIER_AVAILABLE, MQTT_PROP_TYPE_BYTE}, {MQTT_PROP_SHARED_SUBSCRIPTION_AVAILABLE, MQTT_PROP_TYPE_BYTE}}; -void mg_mqtt_send_header(struct mg_connection *c, uint8_t cmd, uint8_t flags, - uint32_t len) { +static bool mqtt_send_header(struct mg_connection *c, uint8_t cmd, + uint8_t flags, uint32_t len) { uint8_t buf[1 + sizeof(len)], *vlen = &buf[1]; buf[0] = (uint8_t) ((cmd << 4) | flags); do { @@ -56,15 +56,20 @@ void mg_mqtt_send_header(struct mg_connection *c, uint8_t cmd, uint8_t flags, if (len > 0) *vlen |= 0x80; vlen++; } while (len > 0 && vlen < &buf[sizeof(buf)]); - mg_send(c, buf, (size_t) (vlen - buf)); + return mg_send(c, buf, (size_t) (vlen - buf)); } -static void mg_send_u16(struct mg_connection *c, uint16_t value) { - mg_send(c, &value, sizeof(value)); +void mg_mqtt_send_header(struct mg_connection *c, uint8_t cmd, uint8_t flags, + uint32_t len) { + if (!mqtt_send_header(c, cmd, flags, len)) mg_error(c, "OOM"); } -static void mg_send_u32(struct mg_connection *c, uint32_t value) { - mg_send(c, &value, sizeof(value)); +static bool mg_send_u16(struct mg_connection *c, uint16_t value) { + return mg_send(c, &value, sizeof(value)); +} + +static bool mg_send_u32(struct mg_connection *c, uint32_t value) { + return mg_send(c, &value, sizeof(value)); } static uint8_t varint_size(size_t length) { @@ -157,46 +162,50 @@ static size_t get_props_size(struct mg_mqtt_prop *props, size_t count) { return size; } -static void mg_send_mqtt_properties(struct mg_connection *c, +static bool mg_send_mqtt_properties(struct mg_connection *c, struct mg_mqtt_prop *props, size_t nprops) { size_t total_size = get_properties_length(props, nprops); uint8_t buf_v[4] = {0, 0, 0, 0}; uint8_t buf[4] = {0, 0, 0, 0}; size_t i, len = encode_varint(buf, total_size); - mg_send(c, buf, (size_t) len); + if (!mg_send(c, buf, (size_t) len)) return false; for (i = 0; i < nprops; i++) { - mg_send(c, &props[i].id, sizeof(props[i].id)); + if (!mg_send(c, &props[i].id, sizeof(props[i].id))) return false; switch (mqtt_prop_type_by_id(props[i].id)) { case MQTT_PROP_TYPE_STRING_PAIR: - mg_send_u16(c, mg_htons((uint16_t) props[i].key.len)); - mg_send(c, props[i].key.buf, props[i].key.len); - mg_send_u16(c, mg_htons((uint16_t) props[i].val.len)); - mg_send(c, props[i].val.buf, props[i].val.len); + if (!mg_send_u16(c, mg_htons((uint16_t) props[i].key.len)) || + !mg_send(c, props[i].key.buf, props[i].key.len) || + !mg_send_u16(c, mg_htons((uint16_t) props[i].val.len)) || + !mg_send(c, props[i].val.buf, props[i].val.len)) + return false; break; case MQTT_PROP_TYPE_BYTE: - mg_send(c, &props[i].iv, sizeof(uint8_t)); + if (!mg_send(c, &props[i].iv, sizeof(uint8_t))) return false; break; case MQTT_PROP_TYPE_SHORT: - mg_send_u16(c, mg_htons((uint16_t) props[i].iv)); + if (!mg_send_u16(c, mg_htons((uint16_t) props[i].iv))) return false; break; case MQTT_PROP_TYPE_INT: - mg_send_u32(c, mg_htonl((uint32_t) props[i].iv)); + if (!mg_send_u32(c, mg_htonl((uint32_t) props[i].iv))) return false; break; case MQTT_PROP_TYPE_STRING: - mg_send_u16(c, mg_htons((uint16_t) props[i].val.len)); - mg_send(c, props[i].val.buf, props[i].val.len); + if (!mg_send_u16(c, mg_htons((uint16_t) props[i].val.len)) || + !mg_send(c, props[i].val.buf, props[i].val.len)) + return false; break; case MQTT_PROP_TYPE_BINARY_DATA: - mg_send_u16(c, mg_htons((uint16_t) props[i].val.len)); - mg_send(c, props[i].val.buf, props[i].val.len); + if (!mg_send_u16(c, mg_htons((uint16_t) props[i].val.len)) || + !mg_send(c, props[i].val.buf, props[i].val.len)) + return false; break; case MQTT_PROP_TYPE_VARIABLE_INT: len = encode_varint(buf_v, props[i].iv); - mg_send(c, buf_v, (size_t) len); + if (!mg_send(c, buf_v, (size_t) len)) return false; break; } } + return true; } size_t mg_mqtt_next_prop(struct mg_mqtt_message *msg, struct mg_mqtt_prop *prop, @@ -289,33 +298,41 @@ void mg_mqtt_login(struct mg_connection *c, const struct mg_mqtt_opts *opts) { total_len += get_props_size(opts->will_props, opts->num_will_props); } - mg_mqtt_send_header(c, MQTT_CMD_CONNECT, 0, (uint32_t) total_len); - mg_send(c, hdr, sizeof(hdr)); // keepalive == 0 means "do not disconnect us!" - mg_send_u16(c, mg_htons((uint16_t) opts->keepalive)); + if (!mqtt_send_header(c, MQTT_CMD_CONNECT, 0, (uint32_t) total_len) || + !mg_send(c, hdr, sizeof(hdr)) || + !mg_send_u16(c, mg_htons((uint16_t) opts->keepalive))) + goto fail; - if (c->is_mqtt5) mg_send_mqtt_properties(c, opts->props, opts->num_props); + if (c->is_mqtt5 && !mg_send_mqtt_properties(c, opts->props, opts->num_props)) + goto fail; - mg_send_u16(c, mg_htons((uint16_t) cid.len)); - mg_send(c, cid.buf, cid.len); + if (!mg_send_u16(c, mg_htons((uint16_t) cid.len)) || + !mg_send(c, cid.buf, cid.len)) + goto fail; if (hdr[7] & MQTT_HAS_WILL) { - if (c->is_mqtt5) - mg_send_mqtt_properties(c, opts->will_props, opts->num_will_props); + if (c->is_mqtt5 && + !mg_send_mqtt_properties(c, opts->will_props, opts->num_will_props)) + goto fail; - mg_send_u16(c, mg_htons((uint16_t) opts->topic.len)); - mg_send(c, opts->topic.buf, opts->topic.len); - mg_send_u16(c, mg_htons((uint16_t) opts->message.len)); - mg_send(c, opts->message.buf, opts->message.len); - } - if (opts->user.len > 0) { - mg_send_u16(c, mg_htons((uint16_t) opts->user.len)); - mg_send(c, opts->user.buf, opts->user.len); - } - if (opts->pass.len > 0) { - mg_send_u16(c, mg_htons((uint16_t) opts->pass.len)); - mg_send(c, opts->pass.buf, opts->pass.len); + if (!mg_send_u16(c, mg_htons((uint16_t) opts->topic.len)) || + !mg_send(c, opts->topic.buf, opts->topic.len) || + !mg_send_u16(c, mg_htons((uint16_t) opts->message.len)) || + !mg_send(c, opts->message.buf, opts->message.len)) + goto fail; } + if (opts->user.len > 0 && + (!mg_send_u16(c, mg_htons((uint16_t) opts->user.len)) || + !mg_send(c, opts->user.buf, opts->user.len))) + goto fail; + if (opts->pass.len > 0 && + (!mg_send_u16(c, mg_htons((uint16_t) opts->pass.len)) || + !mg_send(c, opts->pass.buf, opts->pass.len))) + goto fail; + return; +fail: + mg_error(c, "OOM"); } uint16_t mg_mqtt_pub(struct mg_connection *c, const struct mg_mqtt_opts *opts) { @@ -330,20 +347,28 @@ uint16_t mg_mqtt_pub(struct mg_connection *c, const struct mg_mqtt_opts *opts) { if (c->is_mqtt5) len += get_props_size(opts->props, opts->num_props); if (opts->qos > 0 && id != 0) flags |= 1 << 3; - mg_mqtt_send_header(c, MQTT_CMD_PUBLISH, flags, (uint32_t) len); - mg_send_u16(c, mg_htons((uint16_t) opts->topic.len)); - mg_send(c, opts->topic.buf, opts->topic.len); + if (!mqtt_send_header(c, MQTT_CMD_PUBLISH, flags, (uint32_t) len) || + !mg_send_u16(c, mg_htons((uint16_t) opts->topic.len)) || + !mg_send(c, opts->topic.buf, opts->topic.len)) + goto fail; if (opts->qos > 0) { // need to send 'id' field if (id == 0) { // generate new one if not resending if (++c->mgr->mqtt_id == 0) ++c->mgr->mqtt_id; id = c->mgr->mqtt_id; } - mg_send_u16(c, mg_htons(id)); + if (!mg_send_u16(c, mg_htons(id))) goto fail; } - if (c->is_mqtt5) mg_send_mqtt_properties(c, opts->props, opts->num_props); + if (c->is_mqtt5 && !mg_send_mqtt_properties(c, opts->props, opts->num_props)) + goto fail; - if (opts->message.len > 0) mg_send(c, opts->message.buf, opts->message.len); + if (opts->message.len > 0 && + !mg_send(c, opts->message.buf, opts->message.len)) + goto fail; + return id; + +fail: + mg_error(c, "OOM"); return id; } @@ -352,14 +377,20 @@ void mg_mqtt_sub(struct mg_connection *c, const struct mg_mqtt_opts *opts) { size_t plen = c->is_mqtt5 ? get_props_size(opts->props, opts->num_props) : 0; size_t len = 2 + opts->topic.len + 2 + 1 + plen; - mg_mqtt_send_header(c, MQTT_CMD_SUBSCRIBE, 2, (uint32_t) len); + if (!mqtt_send_header(c, MQTT_CMD_SUBSCRIBE, 2, (uint32_t) len)) goto fail; if (++c->mgr->mqtt_id == 0) ++c->mgr->mqtt_id; - mg_send_u16(c, mg_htons(c->mgr->mqtt_id)); - if (c->is_mqtt5) mg_send_mqtt_properties(c, opts->props, opts->num_props); + if (!mg_send_u16(c, mg_htons(c->mgr->mqtt_id))) goto fail; - mg_send_u16(c, mg_htons((uint16_t) opts->topic.len)); - mg_send(c, opts->topic.buf, opts->topic.len); - mg_send(c, &qos_, sizeof(qos_)); + if (c->is_mqtt5 && !mg_send_mqtt_properties(c, opts->props, opts->num_props)) + goto fail; + + if (!mg_send_u16(c, mg_htons((uint16_t) opts->topic.len)) || + !mg_send(c, opts->topic.buf, opts->topic.len) || + !mg_send(c, &qos_, sizeof(qos_))) + goto fail; + return; +fail: + mg_error(c, "OOM"); } int mg_mqtt_parse(const uint8_t *buf, size_t len, uint8_t version, @@ -466,15 +497,16 @@ static void mqtt_cb(struct mg_connection *c, int ev, void *ev_data) { uint32_t remaining_len = sizeof(id); if (c->is_mqtt5) remaining_len += 2; // 3.4.2 - mg_mqtt_send_header( - c, - (uint8_t) (mm.qos == 2 ? MQTT_CMD_PUBREC : MQTT_CMD_PUBACK), - 0, remaining_len); - mg_send(c, &id, sizeof(id)); + if (!mqtt_send_header(c, + (uint8_t) (mm.qos == 2 ? MQTT_CMD_PUBREC + : MQTT_CMD_PUBACK), + 0, remaining_len) || + !mg_send(c, &id, sizeof(id))) + goto fail; if (c->is_mqtt5) { uint16_t zero = 0; - mg_send(c, &zero, sizeof(zero)); + if (!mg_send(c, &zero, sizeof(zero))) goto fail; } } mg_call(c, MG_EV_MQTT_MSG, &mm); // let the app handle qos stuff @@ -483,15 +515,18 @@ static void mqtt_cb(struct mg_connection *c, int ev, void *ev_data) { case MQTT_CMD_PUBREC: { // MQTT5: 3.5.2-1 TODO(): variable header rc uint16_t id = mg_ntohs(mm.id); uint32_t remaining_len = sizeof(id); // MQTT5 3.6.2-1 - mg_mqtt_send_header(c, MQTT_CMD_PUBREL, 2, remaining_len); - mg_send(c, &id, sizeof(id)); // MQTT5 3.6.1-1, flags = 2 + if (!mqtt_send_header(c, MQTT_CMD_PUBREL, 2, + remaining_len) // MQTT5 3.6.1-1, flags = 2 + || !mg_send(c, &id, sizeof(id))) + goto fail; break; } case MQTT_CMD_PUBREL: { // MQTT5: 3.6.2-1 TODO(): variable header rc uint16_t id = mg_ntohs(mm.id); uint32_t remaining_len = sizeof(id); // MQTT5 3.7.2-1 - mg_mqtt_send_header(c, MQTT_CMD_PUBCOMP, 0, remaining_len); - mg_send(c, &id, sizeof(id)); + if (!mqtt_send_header(c, MQTT_CMD_PUBCOMP, 0, remaining_len) || + !mg_send(c, &id, sizeof(id))) + goto fail; break; } } @@ -503,6 +538,9 @@ static void mqtt_cb(struct mg_connection *c, int ev, void *ev_data) { } } (void) ev_data; + return; +fail: + mg_error(c, "OOM"); } void mg_mqtt_ping(struct mg_connection *nc) { @@ -517,19 +555,24 @@ void mg_mqtt_disconnect(struct mg_connection *c, const struct mg_mqtt_opts *opts) { size_t len = 0; if (c->is_mqtt5) len = 1 + get_props_size(opts->props, opts->num_props); - mg_mqtt_send_header(c, MQTT_CMD_DISCONNECT, 0, (uint32_t) len); + if (!mqtt_send_header(c, MQTT_CMD_DISCONNECT, 0, (uint32_t) len)) goto fail; if (c->is_mqtt5) { uint8_t zero = 0; - mg_send(c, &zero, sizeof(zero)); // reason code - mg_send_mqtt_properties(c, opts->props, opts->num_props); + if (!mg_send(c, &zero, sizeof(zero)) // reason code + || !mg_send_mqtt_properties(c, opts->props, opts->num_props)) + goto fail; } + return; +fail: + mg_error(c, "OOM"); } struct mg_connection *mg_mqtt_connect(struct mg_mgr *mgr, const char *url, const struct mg_mqtt_opts *opts, mg_event_handler_t fn, void *fn_data) { - struct mg_connection *c = mg_connect_svc(mgr, url, fn, fn_data, mqtt_cb, NULL); + struct mg_connection *c = + mg_connect_svc(mgr, url, fn, fn_data, mqtt_cb, NULL); if (c != NULL) { struct mg_mqtt_opts empty; memset(&empty, 0, sizeof(empty)); diff --git a/src/net_builtin.c b/src/net_builtin.c index 0c1acd1e..29a1f6b1 100644 --- a/src/net_builtin.c +++ b/src/net_builtin.c @@ -1924,6 +1924,7 @@ bool mg_open_listener(struct mg_connection *c, const char *url) { static void write_conn(struct mg_connection *c) { long len = c->is_tls ? mg_tls_send(c, c->send.buf, c->send.len) : mg_io_send(c, c->send.buf, c->send.len); + // TODO(): mg_tls_send() may return 0 forever on steady OOM if (len == MG_IO_ERR) { mg_error(c, "tx err"); } else if (len > 0) { @@ -1936,14 +1937,8 @@ static void init_closure(struct mg_connection *c) { struct connstate *s = (struct connstate *) (c + 1); if (c->is_udp == false && c->is_listening == false && c->is_connecting == false) { // For TCP conns, -<<<<<<< HEAD tx_tcp(c->mgr->ifp, s->mac, &c->loc, &c->rem, TH_FIN | TH_ACK, mg_htonl(s->seq), mg_htonl(s->ack), NULL, 0); -======= - uint32_t rem_ip = c->rem.ip4; - tx_tcp(c->mgr->ifp, s->mac, rem_ip, TH_FIN | TH_ACK, c->loc.port, - c->rem.port, mg_htonl(s->seq), mg_htonl(s->ack), NULL, 0); ->>>>>>> 81b87991 (wip) settmout(c, MIP_TTYPE_FIN); } } @@ -2009,7 +2004,7 @@ bool mg_send(struct mg_connection *c, const void *buf, size_t len) { len = trim_len(c, len); // Trimming length if necessary res = tx_udp(ifp, s->mac, &c->loc, &c->rem, buf, len); } else { - res = mg_iobuf_add(&c->send, c->send.len, buf, len); + res = (bool) mg_iobuf_add(&c->send, c->send.len, buf, len); // res == 0 means an OOM condition (iobuf couldn't resize), yet this is so // far recoverable, let the caller decide } diff --git a/src/sock.c b/src/sock.c index 5e87d1fa..0c664548 100644 --- a/src/sock.c +++ b/src/sock.c @@ -141,7 +141,7 @@ bool mg_send(struct mg_connection *c, const void *buf, size_t len) { iolog(c, (char *) buf, n, false); return n > 0; } else { - return mg_iobuf_add(&c->send, c->send.len, buf, len); + return (bool) mg_iobuf_add(&c->send, c->send.len, buf, len); // returning 0 means an OOM condition (iobuf couldn't resize), yet this is // so far recoverable, let the caller decide } @@ -341,6 +341,7 @@ static void write_conn(struct mg_connection *c) { char *buf = (char *) c->send.buf; size_t len = c->send.len; long n = c->is_tls ? mg_tls_send(c, buf, len) : mg_io_send(c, buf, len); + // TODO(): mg_tls_send() may return 0 forever on steady OOM MG_DEBUG(("%lu %ld snd %ld/%ld rcv %ld/%ld n=%ld err=%d", c->id, c->fd, (long) c->send.len, (long) c->send.size, (long) c->recv.len, (long) c->recv.size, n, MG_SOCK_ERR(n))); diff --git a/src/ws.c b/src/ws.c index be86af49..cc5063ba 100644 --- a/src/ws.c +++ b/src/ws.c @@ -55,7 +55,7 @@ static void ws_handshake(struct mg_connection *c, const struct mg_str *wskey, mg_printf(c, "Sec-WebSocket-Protocol: %.*s\r\n", (int) wsproto->len, wsproto->buf); } - mg_send(c, "\r\n", 2); + if (!mg_send(c, "\r\n", 2)) mg_error(c, "OOM"); } static uint32_t be32(const uint8_t *p) {