mirror of https://github.com/arendst/Tasmota.git
Add more checks to MQTT file upload
This commit is contained in:
Theo Arends
parent
ea83c8acd1
commit
dd624c3ce6
|
|
@ -23,11 +23,14 @@
|
|||
/*********************************************************************************************\
|
||||
* MQTT file transfer
|
||||
*
|
||||
* Supports base64 encoded binary data transfer
|
||||
* Supports both binary and base64 encoded binary data transfer
|
||||
\*********************************************************************************************/
|
||||
|
||||
#include <PubSubClient.h>
|
||||
#include <base64.hpp>
|
||||
|
||||
extern PubSubClient MqttClient;
|
||||
|
||||
struct FMQTT {
|
||||
uint32_t file_pos = 0; // MQTT file position during upload/download
|
||||
uint32_t file_size = 0; // MQTT total file size
|
||||
|
|
@ -40,10 +43,6 @@ struct FMQTT {
|
|||
uint8_t file_id = 0; // MQTT unique file id during upload/download
|
||||
} FMqtt;
|
||||
|
||||
void MqttTopicSize(uint32_t topic_size) {
|
||||
FMqtt.topic_size = topic_size +1;
|
||||
}
|
||||
|
||||
/*
|
||||
The download chunk size is the data size before it is encoded to base64.
|
||||
It is smaller than the upload chunksize as it is bound by MESSZ
|
||||
|
|
@ -55,18 +54,20 @@ const uint32_t mqtt_file_chuck_size = (((MESSZ - FileTransferHeaderSize) / 4) *
|
|||
|
||||
uint32_t FileUploadChunckSize(void) {
|
||||
/*
|
||||
The upload chunk size is the data size before it is encoded to base64.
|
||||
The upload chunk size is the data size of the payload.
|
||||
It can be larger than the download chunksize which is bound by MESSZ
|
||||
The PubSubClient upload buffer with length MQTT_MAX_PACKET_SIZE (1200) contains
|
||||
- Header of 5 bytes (MQTT_MAX_HEADER_SIZE)
|
||||
- Topic string terminated with a zero (stat/demo/FILEUPLOAD<null>)
|
||||
- Payload ({"Id":116,"Data":"<base64 encoded FileUploadChunckSize>"}<null>)
|
||||
- Payload ({"Id":116,"Data":"<base64 encoded FileUploadChunckSize>"}<null>) or (<binary data>)
|
||||
*/
|
||||
const uint32_t PubSubClientHeaderSize = 5; // MQTT_MAX_HEADER_SIZE
|
||||
return MQTT_MAX_PACKET_SIZE - PubSubClientHeaderSize - FMqtt.topic_size - FileTransferHeaderSize;
|
||||
return MqttClient.getBufferSize() - PubSubClientHeaderSize - FMqtt.topic_size -1;
|
||||
}
|
||||
|
||||
uint32_t MqttFileUploadValidate(uint32_t rcv_id) {
|
||||
if (XdrvMailbox.grpflg) { return 5; }
|
||||
|
||||
if ((0 == FMqtt.file_id) && (rcv_id > 0) && (FMqtt.file_size > 0) && (FMqtt.file_type > 0)) {
|
||||
FMqtt.file_buffer = nullptr; // Init upload buffer
|
||||
|
||||
|
|
@ -74,7 +75,12 @@ uint32_t MqttFileUploadValidate(uint32_t rcv_id) {
|
|||
return 1; // Invalid password
|
||||
}
|
||||
|
||||
if (UPL_SETTINGS != FMqtt.file_type) { // Check enough flash space for intermediate upload
|
||||
// Check buffer size
|
||||
if (UPL_SETTINGS == FMqtt.file_type) {
|
||||
if (FMqtt.file_size > 4096) {
|
||||
return 2; // Settings supports max 4k size
|
||||
}
|
||||
} else { // Check enough flash space for intermediate upload
|
||||
uint32_t head_room = (FlashWriteMaxSector() - FlashWriteStartSector()) * SPI_FLASH_SEC_SIZE;
|
||||
uint32_t rounded_size = (FMqtt.file_size + SPI_FLASH_SEC_SIZE -1) & (~(SPI_FLASH_SEC_SIZE - 1));
|
||||
if (rounded_size > head_room) {
|
||||
|
|
@ -82,9 +88,11 @@ uint32_t MqttFileUploadValidate(uint32_t rcv_id) {
|
|||
}
|
||||
}
|
||||
|
||||
// Init file_buffer
|
||||
if (UPL_TASMOTA == FMqtt.file_type) {
|
||||
if (Update.begin(FMqtt.file_size)) {
|
||||
FMqtt.file_buffer = &FMqtt.file_id; // Dummy buffer
|
||||
// TasmotaGlobal.blinkstate = true; // Stay lit
|
||||
SettingsSave(1); // Free flash for OTA update
|
||||
}
|
||||
}
|
||||
|
|
@ -122,24 +130,31 @@ uint32_t MqttFileUploadValidate(uint32_t rcv_id) {
|
|||
|
||||
void CmndFileUpload(void) {
|
||||
/*
|
||||
Upload (binary) max 700 bytes chunks of data base64 encoded with MD5 hash over base64 decoded data
|
||||
FileUpload 0 - Abort current upload
|
||||
FileUpload {"File":"Config_wemos10_9.4.0.3.dmp","Id":116,"Type":2,"Size":4096}
|
||||
FileUpload {"Id":116,"Data":"CRJcTQ9fYGF ... OT1BRUlNUVVZXWFk="}
|
||||
FileUpload {"Id":116,"Data":" ... "}
|
||||
FileUpload {"Id":116,"Md5":"496fcbb433bbca89833063174d2c5747"}
|
||||
*/
|
||||
if (XdrvMailbox.grpflg) { return; }
|
||||
Upload <MaxSize> bytes chunks of data either base64 encoded or binary with MD5 hash
|
||||
|
||||
Supported Type:
|
||||
1 - OTA firmware
|
||||
2 - Settings
|
||||
|
||||
FileUpload 0 - Abort current upload
|
||||
|
||||
Start an upload session:
|
||||
FileUpload {"Password":"","File":"Config_wemos10_9.4.0.3.dmp","Id":116,"Type":2,"Size":4096}
|
||||
|
||||
Upload data using base64:
|
||||
FileUpload {"Id":116,"Data":"CRJcTQ9fYGF ... OT1BRUlNUVVZXWFk="}
|
||||
FileUpload {"Id":116,"Data":" ... "}
|
||||
Or binary:
|
||||
FileUpload201 <binary data>
|
||||
|
||||
Finish upload session:
|
||||
FileUpload {"Id":116,"Md5":"496fcbb433bbca89833063174d2c5747"}
|
||||
*/
|
||||
const char* base64_data = nullptr;
|
||||
uint32_t rcv_id = 0;
|
||||
char* dataBuf = (char*)XdrvMailbox.data;
|
||||
|
||||
bool binary_data = false;
|
||||
if (XdrvMailbox.index > 199) { // Check for raw data
|
||||
XdrvMailbox.index -= 200;
|
||||
binary_data = true;
|
||||
}
|
||||
bool binary_data = (XdrvMailbox.index > 199); // Check for raw data
|
||||
|
||||
if (!binary_data) {
|
||||
if (strlen(dataBuf) > 8) { // Workaround exception if empty JSON like {} - Needs checks
|
||||
|
|
@ -156,7 +171,7 @@ void CmndFileUpload(void) {
|
|||
if (val) { FMqtt.file_md5 = val.getStr(); }
|
||||
val = root[PSTR("DATA")];
|
||||
if (val) { base64_data = val.getStr(); }
|
||||
val = root[PSTR("PASS")];
|
||||
val = root[PSTR("PASSWORD")];
|
||||
if (val) { FMqtt.file_password = val.getStr(); }
|
||||
}
|
||||
}
|
||||
|
|
@ -170,7 +185,7 @@ void CmndFileUpload(void) {
|
|||
|
||||
TasmotaGlobal.masterlog_level = LOG_LEVEL_NONE; // Enable logging
|
||||
|
||||
char error_txt[TOPSZ];
|
||||
char error_txt[20];
|
||||
snprintf_P(error_txt, sizeof(error_txt), PSTR(D_JSON_ERROR " %d"), error);
|
||||
ResponseCmndChar(error_txt);
|
||||
}
|
||||
|
|
@ -209,7 +224,7 @@ void CmndFileUpload(void) {
|
|||
|
||||
if ((FMqtt.file_pos > rcvd_bytes) && ((FMqtt.file_pos % 102400) <= rcvd_bytes)) {
|
||||
TasmotaGlobal.masterlog_level = LOG_LEVEL_NONE; // Enable logging
|
||||
AddLog(LOG_LEVEL_DEBUG, PSTR(D_LOG_UPLOAD "Progress %d kB"), FMqtt.file_pos / 1024);
|
||||
AddLog(LOG_LEVEL_DEBUG, PSTR(D_LOG_UPLOAD "Progress %d kB"), (FMqtt.file_pos / 10240) * 10);
|
||||
TasmotaGlobal.masterlog_level = LOG_LEVEL_DEBUG_MORE; // Hide upload data logging
|
||||
}
|
||||
}
|
||||
|
|
@ -219,7 +234,7 @@ void CmndFileUpload(void) {
|
|||
|
||||
uint32_t chunk_size = FileUploadChunckSize();
|
||||
if (!binary_data) {
|
||||
chunk_size = ((chunk_size / 4) * 3) -2; // Calculate base64 chunk size
|
||||
chunk_size = (((chunk_size - FileTransferHeaderSize) / 4) * 3) -2; // Calculate base64 chunk size
|
||||
}
|
||||
// {"Id":116,"MaxSize":"765"}
|
||||
Response_P(PSTR("{\"Id\":%d,\"MaxSize\":%d}"), FMqtt.file_id, chunk_size);
|
||||
|
|
|
|||
|
|
@ -522,6 +522,12 @@ void MqttDataHandler(char* mqtt_topic, uint8_t* mqtt_data, unsigned int data_len
|
|||
}
|
||||
}
|
||||
|
||||
#ifdef USE_MQTT_FILE
|
||||
FMqtt.topic_size = strlen(mqtt_topic);
|
||||
#endif // USE_MQTT_FILE
|
||||
|
||||
// AddLog(LOG_LEVEL_DEBUG, PSTR(D_LOG_MQTT "BufferSize %d, Topic |%s|, Length %d, data_len %d"), MqttClient.getBufferSize(), mqtt_topic, strlen(mqtt_topic), data_len);
|
||||
|
||||
// Save MQTT data ASAP as it's data is discarded by PubSubClient with next publish as used in MQTTlog
|
||||
char topic[TOPSZ];
|
||||
#ifdef USE_MQTT_AZURE_IOT
|
||||
|
|
@ -544,9 +550,6 @@ void MqttDataHandler(char* mqtt_topic, uint8_t* mqtt_data, unsigned int data_len
|
|||
#else
|
||||
strlcpy(topic, mqtt_topic, sizeof(topic));
|
||||
#endif // USE_MQTT_AZURE_IOT
|
||||
#ifdef USE_MQTT_FILE
|
||||
MqttTopicSize(strlen(topic));
|
||||
#endif // USE_MQTT_FILE
|
||||
mqtt_data[data_len] = 0;
|
||||
char data[data_len +1];
|
||||
memcpy(data, mqtt_data, sizeof(data));
|
||||
|
|
|
|||
Binary file not shown.
|
|
@ -1,142 +0,0 @@
|
|||
#!/usr/bin/env python
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
"""
|
||||
upload-ota-201.py - Upload Tasmota firmware file
|
||||
|
||||
Copyright (C) 2021 Theo Arends
|
||||
|
||||
This program is free software: you can redistribute it and/or modify
|
||||
it under the terms of the GNU General Public License as published by
|
||||
the Free Software Foundation, either version 3 of the License, or
|
||||
(at your option) any later version.
|
||||
|
||||
This program is distributed in the hope that it will be useful,
|
||||
but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
GNU General Public License for more details.
|
||||
|
||||
You should have received a copy of the GNU General Public License
|
||||
along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
Requirements:
|
||||
- Python 3.x and Pip:
|
||||
sudo apt-get install python3 python3-pip
|
||||
pip3 install paho-mqtt json
|
||||
|
||||
Instructions:
|
||||
Edit file and change parameters in User Configuration Section
|
||||
|
||||
Then execute command upload-ota-201.py
|
||||
|
||||
"""
|
||||
|
||||
import paho.mqtt.client as mqtt
|
||||
import time
|
||||
import base64
|
||||
import hashlib
|
||||
import json
|
||||
|
||||
# **** Start of User Configuration Section
|
||||
|
||||
broker = "domus1" # MQTT broker ip address or name
|
||||
broker_port = 1883 # MQTT broker port
|
||||
|
||||
mytopic = "demo" # Tasmota MQTT topic
|
||||
#myfile = "../../build_output/firmware/tasmota32.bin" # Tasmota esp32 firmware file name
|
||||
myfile = "../../build_output/firmware/tasmota.bin.gz" # Tasmota esp8266 firmware file name
|
||||
myfiletype = 1 # Tasmota firmware file type
|
||||
|
||||
# **** End of User Configuration Section
|
||||
|
||||
# Derive fulltopic from broker LWT message
|
||||
mypublish = "cmnd/"+mytopic+"/fileupload"
|
||||
mysubscribe = "stat/"+mytopic+"/FILEUPLOAD" # Case sensitive
|
||||
|
||||
Ack_flag = False
|
||||
|
||||
file_id = 114 # Even id between 2 and 254
|
||||
file_chunk_size = 700 # Default Tasmota MQTT max message size
|
||||
|
||||
# The callback for when mysubscribe message is received
|
||||
def on_message(client, userdata, msg):
|
||||
global Ack_flag
|
||||
global file_chunk_size
|
||||
|
||||
rcv_code = ""
|
||||
rcv_id = 0
|
||||
|
||||
# print("Received message =",str(msg.payload.decode("utf-8")))
|
||||
|
||||
root = json.loads(msg.payload.decode("utf-8"))
|
||||
if "Command" in root: rcv_code = root["Command"]
|
||||
if rcv_code == "Error":
|
||||
print("Error: Command error")
|
||||
return
|
||||
|
||||
if "Id" in root: rcv_id = root["Id"]
|
||||
if rcv_id == file_id:
|
||||
if "MaxSize" in root: file_chunk_size = root["MaxSize"]
|
||||
|
||||
Ack_flag = False
|
||||
|
||||
def wait_for_ack():
|
||||
global Ack_flag
|
||||
timeout = 100
|
||||
while Ack_flag and timeout > 0:
|
||||
time.sleep(0.01)
|
||||
timeout = timeout -1
|
||||
|
||||
if Ack_flag:
|
||||
print("Error: Ack timeout")
|
||||
|
||||
return Ack_flag
|
||||
|
||||
client = mqtt.Client()
|
||||
client.on_message = on_message
|
||||
client.connect(broker, broker_port)
|
||||
client.loop_start() # Start loop to process received messages
|
||||
client.subscribe(mysubscribe)
|
||||
|
||||
time_start = time.time()
|
||||
print("Uploading file "+myfile+" to "+mytopic+" ...")
|
||||
|
||||
fo = open(myfile,"rb")
|
||||
fo.seek(0, 2) # os.SEEK_END
|
||||
file_size = fo.tell()
|
||||
fo.seek(0, 0) # os.SEEK_SET
|
||||
|
||||
client.publish(mypublish, "{\"File\":\""+myfile+"\",\"Id\":"+str("%3d"%file_id)+",\"Type\":"+str(myfiletype)+",\"Size\":"+str(file_size)+"}")
|
||||
Ack_flag = True
|
||||
|
||||
out_hash_md5 = hashlib.md5()
|
||||
|
||||
Run_flag = True
|
||||
while Run_flag:
|
||||
if wait_for_ack(): # We use Ack here
|
||||
Run_flag = False
|
||||
|
||||
else:
|
||||
chunk = fo.read(file_chunk_size)
|
||||
if chunk:
|
||||
out_hash_md5.update(chunk) # Update hash
|
||||
|
||||
# base64_encoded_data = base64.b64encode(chunk)
|
||||
# base64_data = base64_encoded_data.decode('utf-8')
|
||||
# Message length used by Tasmota (FileTransferHeaderSize)
|
||||
# client.publish(mypublish, "{\"Id\":"+str("%3d"%file_id)+",\"Data\":\""+base64_data+"\"}")
|
||||
client.publish(mypublish+"201", chunk)
|
||||
Ack_flag = True
|
||||
|
||||
else:
|
||||
md5_hash = out_hash_md5.hexdigest()
|
||||
client.publish(mypublish, "{\"Id\":"+str("%3d"%file_id)+",\"Md5\":\""+md5_hash+"\"}")
|
||||
Run_flag = False
|
||||
|
||||
fo.close()
|
||||
|
||||
time_taken = time.time() - time_start
|
||||
print("Done in "+str("%.2f"%time_taken)+" seconds")
|
||||
|
||||
client.disconnect() # Disconnect
|
||||
client.loop_stop() # Stop loop
|
||||
|
|
@ -113,7 +113,7 @@ fo.seek(0, 2) # os.SEEK_END
|
|||
file_size = fo.tell()
|
||||
fo.seek(0, 0) # os.SEEK_SET
|
||||
|
||||
client.publish(mypublish, "{\"Pass\":\""+mypassword+"\",\"File\":\""+myfile+"\",\"Id\":"+str("%3d"%file_id)+",\"Type\":"+str(myfiletype)+",\"Size\":"+str(file_size)+"}")
|
||||
client.publish(mypublish, "{\"Password\":\""+mypassword+"\",\"File\":\""+myfile+"\",\"Id\":"+str("%3d"%file_id)+",\"Type\":"+str(myfiletype)+",\"Size\":"+str(file_size)+"}")
|
||||
Ack_flag = True
|
||||
|
||||
out_hash_md5 = hashlib.md5()
|
||||
|
|
|
|||
|
|
@ -112,7 +112,7 @@ fo.seek(0, 2) # os.SEEK_END
|
|||
file_size = fo.tell()
|
||||
fo.seek(0, 0) # os.SEEK_SET
|
||||
|
||||
client.publish(mypublish, "{\"Pass\":\""+mypassword+"\",\"File\":\""+myfile+"\",\"Id\":"+str("%3d"%file_id)+",\"Type\":"+str(myfiletype)+",\"Size\":"+str(file_size)+"}")
|
||||
client.publish(mypublish, "{\"Password\":\""+mypassword+"\",\"File\":\""+myfile+"\",\"Id\":"+str("%3d"%file_id)+",\"Type\":"+str(myfiletype)+",\"Size\":"+str(file_size)+"}")
|
||||
Ack_flag = True
|
||||
|
||||
out_hash_md5 = hashlib.md5()
|
||||
|
|
|
|||
Loading…
Reference in New Issue