mh
/
mhsw-espurna
1
0
Fork 0
Code Releases 0 Activity
Fork of the espurna firmware for `mhsw` switches
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
2858 Commits
18 Branches
212 MiB
 
 
 
 
 
 
Tree: 2ecce4d136
Branches Tags
${ item.name }
Create branch ${ searchTerm }
from '2ecce4d136'
${ noResults }
mhsw-espurna/code/espurna/ws.ino

813 lines
22 KiB
Raw Blame History

/*
WEBSOCKET MODULE
Copyright (C) 2016-2019 by Xose Pérez <xose dot perez at gmail dot com>
*/
#if WEB_SUPPORT
#include <ESPAsyncTCP.h>
#include <ESPAsyncWebServer.h>
#include <ArduinoJson.h>
#include <Ticker.h>
#include <vector>
#include "libs/WebSocketIncommingBuffer.h"
AsyncWebSocket _ws("/ws");
Ticker _ws_defer;
uint32_t _ws_last_update = 0;
// -----------------------------------------------------------------------------
// WS callbacks
// -----------------------------------------------------------------------------
ws_callbacks_t& ws_callbacks_t::onVisible(ws_on_send_callback_f cb) {
on_visible.push_back(cb);
return *this;
}
ws_callbacks_t& ws_callbacks_t::onConnected(ws_on_send_callback_f cb) {
on_connected.push_back(cb);
return *this;
}
ws_callbacks_t& ws_callbacks_t::onData(ws_on_send_callback_f cb) {
on_data.push_back(cb);
return *this;
}
ws_callbacks_t& ws_callbacks_t::onAction(ws_on_action_callback_f cb) {
on_action.push_back(cb);
return *this;
}
ws_callbacks_t& ws_callbacks_t::onKeyCheck(ws_on_keycheck_callback_f cb) {
on_keycheck.push_back(cb);
return *this;
}
ws_callbacks_t _ws_callbacks;
struct ws_counter_t {
ws_counter_t() : current(0), start(0), stop(0) {}
ws_counter_t(uint32_t start, uint32_t stop) :
current(start), start(start), stop(stop) {}
void reset() {
current = start;
}
void next() {
if (current < stop) {
++current;
}
}
bool done() {
return (current >= stop);
}
uint32_t current;
uint32_t start;
uint32_t stop;
};
struct ws_data_t {
enum mode_t {
SEQUENCE,
ALL
};
ws_data_t(const ws_on_send_callback_f& cb) :
storage(new ws_on_send_callback_list_t {cb}),
client_id(0),
mode(ALL),
callbacks(*storage.get()),
counter(0, 1)
{}
ws_data_t(uint32_t client_id, const ws_on_send_callback_f& cb) :
storage(new ws_on_send_callback_list_t {cb}),
client_id(client_id),
mode(ALL),
callbacks(*storage.get()),
counter(0, 1)
{}
ws_data_t(const uint32_t client_id, ws_on_send_callback_list_t&& callbacks, mode_t mode = SEQUENCE) :
storage(new ws_on_send_callback_list_t(std::move(callbacks))),
client_id(client_id),
mode(mode),
callbacks(*storage.get()),
counter(0, (storage.get())->size())
{}
ws_data_t(const uint32_t client_id, const ws_on_send_callback_list_t& callbacks, mode_t mode = SEQUENCE) :
client_id(client_id),
mode(mode),
callbacks(callbacks),
counter(0, callbacks.size())
{}
bool done() {
return counter.done();
}
void sendAll(JsonObject& root) {
while (!counter.done()) counter.next();
for (auto& callback : callbacks) {
callback(root);
}
}
void sendCurrent(JsonObject& root) {
callbacks[counter.current](root);
counter.next();
}
void send(JsonObject& root) {
switch (mode) {
case SEQUENCE: sendCurrent(root); break;
case ALL: sendAll(root); break;
}
}
std::unique_ptr<ws_on_send_callback_list_t> storage;
const uint32_t client_id;
const mode_t mode;
const ws_on_send_callback_list_t& callbacks;
ws_counter_t counter;
};
std::queue<ws_data_t> _ws_client_data;
// -----------------------------------------------------------------------------
// WS authentication
// -----------------------------------------------------------------------------
struct ws_ticket_t {
IPAddress ip;
unsigned long timestamp = 0;
};
ws_ticket_t _ws_tickets[WS_BUFFER_SIZE];
void _onAuth(AsyncWebServerRequest *request) {
webLog(request);
if (!webAuthenticate(request)) return request->requestAuthentication();
IPAddress ip = request->client()->remoteIP();
unsigned long now = millis();
unsigned short index;
for (index = 0; index < WS_BUFFER_SIZE; index++) {
if (_ws_tickets[index].ip == ip) break;
if (_ws_tickets[index].timestamp == 0) break;
if (now - _ws_tickets[index].timestamp > WS_TIMEOUT) break;
}
if (index == WS_BUFFER_SIZE) {
request->send(429);
} else {
_ws_tickets[index].ip = ip;
_ws_tickets[index].timestamp = now;
request->send(200, "text/plain", "OK");
}
}
bool _wsAuth(AsyncWebSocketClient * client) {
IPAddress ip = client->remoteIP();
unsigned long now = millis();
unsigned short index = 0;
for (index = 0; index < WS_BUFFER_SIZE; index++) {
if ((_ws_tickets[index].ip == ip) && (now - _ws_tickets[index].timestamp < WS_TIMEOUT)) break;
}
if (index == WS_BUFFER_SIZE) {
return false;
}
return true;
}
// -----------------------------------------------------------------------------
// Debug
// -----------------------------------------------------------------------------
#if DEBUG_WEB_SUPPORT
struct ws_debug_msg_t {
ws_debug_msg_t(const char* prefix, const char* message) :
prefix(prefix), message(message)
{}
String prefix;
String message;
};
struct ws_debug_t {
ws_debug_t(size_t capacity) :
flush(false),
current(0),
capacity(capacity)
{
messages.reserve(capacity);
}
void clear() {
messages.clear();
current = 0;
flush = false;
}
void add(const char* prefix, const char* message) {
if (current >= capacity) {
flush = true;
send(wsConnected());
}
messages.emplace(messages.begin() + current, prefix, message);
flush = true;
++current;
}
void send(const bool connected) {
if (!connected && flush) {
clear();
return;
}
if (!flush) return;
// ref: http://arduinojson.org/v5/assistant/
// {"weblog": {"msg":[...],"pre":[...]}}
DynamicJsonBuffer jsonBuffer(2*JSON_ARRAY_SIZE(messages.size()) + JSON_OBJECT_SIZE(1) + JSON_OBJECT_SIZE(2));
JsonObject& root = jsonBuffer.createObject();
JsonObject& weblog = root.createNestedObject("weblog");
JsonArray& msg = weblog.createNestedArray("msg");
JsonArray& pre = weblog.createNestedArray("pre");
for (auto& message : messages) {
pre.add(message.prefix.c_str());
msg.add(message.message.c_str());
}
wsSend(root);
clear();
}
bool flush;
size_t current;
const size_t capacity;
std::vector<ws_debug_msg_t> messages;
};
// TODO: move to the headers?
constexpr const size_t WS_DEBUG_MSG_BUFFER = 8;
ws_debug_t _ws_debug(WS_DEBUG_MSG_BUFFER);
bool wsDebugSend(const char* prefix, const char* message) {
if (!wsConnected()) return false;
_ws_debug.add(prefix, message);
return true;
}
#endif
// Check the existing setting before saving it
// TODO: this should know of the default values, somehow?
// TODO: move webPort handling somewhere else?
bool _wsStore(const String& key, const String& value) {
if (key == "webPort") {
if ((value.toInt() == 0) || (value.toInt() == 80)) {
return delSetting(key);
}
}
if (value != getSetting(key)) {
return setSetting(key, value);
}
return false;
}
// -----------------------------------------------------------------------------
// Store indexed key (key0, key1, etc.) from array
// -----------------------------------------------------------------------------
bool _wsStore(const String& key, JsonArray& value) {
bool changed = false;
unsigned char index = 0;
for (auto element : value) {
if (_wsStore(key + index, element.as<String>())) changed = true;
index++;
}
// Delete further values
for (unsigned char i=index; i<SETTINGS_MAX_LIST_COUNT; i++) {
if (!delSetting(key, index)) break;
changed = true;
}
return changed;
}
bool _wsCheckKey(const String& key, JsonVariant& value) {
for (auto& callback : _ws_callbacks.on_keycheck) {
if (callback(key.c_str(), value)) return true;
// TODO: remove this to call all OnKeyCheckCallbacks with the
// current key/value
}
return false;
}
void _wsParse(AsyncWebSocketClient *client, uint8_t * payload, size_t length) {
//DEBUG_MSG_P(PSTR("[WEBSOCKET] Parsing: %s\n"), length ? (char*) payload : "");
// Get client ID
uint32_t client_id = client->id();
// Check early for empty object / nothing
if ((length == 0) || (length == 1)) {
return;
}
if ((length == 3) && (strcmp((char*) payload, "{}") == 0)) {
return;
}
// Parse JSON input
// TODO: json buffer should be pretty efficient with the non-const payload,
// most of the space is taken by the object key references
DynamicJsonBuffer jsonBuffer(512);
JsonObject& root = jsonBuffer.parseObject((char *) payload);
if (!root.success()) {
DEBUG_MSG_P(PSTR("[WEBSOCKET] JSON parsing error\n"));
wsSend_P(client_id, PSTR("{\"message\": 3}"));
return;
}
// Check actions -----------------------------------------------------------
const char* action = root["action"];
if (action) {
DEBUG_MSG_P(PSTR("[WEBSOCKET] Requested action: %s\n"), action);
if (strcmp(action, "reboot") == 0) {
deferredReset(100, CUSTOM_RESET_WEB);
return;
}
if (strcmp(action, "reconnect") == 0) {
_ws_defer.once_ms(100, wifiDisconnect);
return;
}
if (strcmp(action, "factory_reset") == 0) {
DEBUG_MSG_P(PSTR("\n\nFACTORY RESET\n\n"));
resetSettings();
deferredReset(100, CUSTOM_RESET_FACTORY);
return;
}
JsonObject& data = root["data"];
if (data.success()) {
// Callbacks
for (auto& callback : _ws_callbacks.on_action) {
callback(client_id, action, data);
}
// Restore configuration via websockets
if (strcmp(action, "restore") == 0) {
if (settingsRestoreJson(data)) {
wsSend_P(client_id, PSTR("{\"message\": 5}"));
} else {
wsSend_P(client_id, PSTR("{\"message\": 4}"));
}
}
return;
}
};
// Check configuration -----------------------------------------------------
JsonObject& config = root["config"];
if (config.success()) {
DEBUG_MSG_P(PSTR("[WEBSOCKET] Parsing configuration data\n"));
String adminPass;
bool save = false;
for (auto kv: config) {
bool changed = false;
String key = kv.key;
JsonVariant& value = kv.value;
// Check password
if (key == "adminPass") {
if (!value.is<JsonArray&>()) continue;
JsonArray& values = value.as<JsonArray&>();
if (values.size() != 2) continue;
if (values[0].as<String>().equals(values[1].as<String>())) {
String password = values[0].as<String>();
if (password.length() > 0) {
setSetting(key, password);
save = true;
wsSend_P(client_id, PSTR("{\"action\": \"reload\"}"));
}
} else {
wsSend_P(client_id, PSTR("{\"message\": 7}"));
}
continue;
}
if (!_wsCheckKey(key, value)) {
delSetting(key);
continue;
}
// Store values
if (value.is<JsonArray&>()) {
if (_wsStore(key, value.as<JsonArray&>())) changed = true;
} else {
if (_wsStore(key, value.as<String>())) changed = true;
}
// Update flags if value has changed
if (changed) {
save = true;
}
}
// Save settings
if (save) {
// Callbacks
espurnaReload();
// Persist settings
saveSettings();
wsSend_P(client_id, PSTR("{\"message\": 8}"));
} else {
wsSend_P(client_id, PSTR("{\"message\": 9}"));
}
}
}
void _wsUpdate(JsonObject& root) {
root["heap"] = getFreeHeap();
root["uptime"] = getUptime();
root["rssi"] = WiFi.RSSI();
root["loadaverage"] = systemLoadAverage();
#if ADC_MODE_VALUE == ADC_VCC
root["vcc"] = ESP.getVcc();
#endif
#if NTP_SUPPORT
if (ntpSynced()) root["now"] = now();
#endif
}
void _wsResetUpdateTimer() {
_ws_last_update = millis() + WS_UPDATE_INTERVAL;
}
void _wsDoUpdate(const bool connected) {
if (!connected) return;
if (millis() - _ws_last_update > WS_UPDATE_INTERVAL) {
_ws_last_update = millis();
wsSend(_wsUpdate);
}
}
bool _wsOnKeyCheck(const char * key, JsonVariant& value) {
if (strncmp(key, "ws", 2) == 0) return true;
if (strncmp(key, "admin", 5) == 0) return true;
if (strncmp(key, "hostname", 8) == 0) return true;
if (strncmp(key, "desc", 4) == 0) return true;
if (strncmp(key, "webPort", 7) == 0) return true;
return false;
}
void _wsOnConnected(JsonObject& root) {
char chipid[7];
snprintf_P(chipid, sizeof(chipid), PSTR("%06X"), ESP.getChipId());
root["webMode"] = WEB_MODE_NORMAL;
root["app_name"] = APP_NAME;
root["app_version"] = APP_VERSION;
root["app_build"] = buildTime();
#if defined(APP_REVISION)
root["app_revision"] = APP_REVISION;
#endif
root["manufacturer"] = MANUFACTURER;
root["chipid"] = String(chipid);
root["mac"] = WiFi.macAddress();
root["bssid"] = WiFi.BSSIDstr();
root["channel"] = WiFi.channel();
root["device"] = DEVICE;
root["hostname"] = getSetting("hostname");
root["desc"] = getSetting("desc");
root["network"] = getNetwork();
root["deviceip"] = getIP();
root["sketch_size"] = ESP.getSketchSize();
root["free_size"] = ESP.getFreeSketchSpace();
root["sdk"] = ESP.getSdkVersion();
root["core"] = getCoreVersion();
root["btnDelay"] = getSetting("btnDelay", BUTTON_DBLCLICK_DELAY).toInt();
root["webPort"] = getSetting("webPort", WEB_PORT).toInt();
root["wsAuth"] = getSetting("wsAuth", WS_AUTHENTICATION).toInt() == 1;
root["hbMode"] = getSetting("hbMode", HEARTBEAT_MODE).toInt();
root["hbInterval"] = getSetting("hbInterval", HEARTBEAT_INTERVAL).toInt();
}
void wsSend(JsonObject& root) {
// TODO: avoid serializing twice?
size_t len = root.measureLength();
AsyncWebSocketMessageBuffer* buffer = _ws.makeBuffer(len);
if (buffer) {
root.printTo(reinterpret_cast<char*>(buffer->get()), len + 1);
_ws.textAll(buffer);
}
}
void wsSend(uint32_t client_id, JsonObject& root) {
AsyncWebSocketClient* client = _ws.client(client_id);
if (client == nullptr) return;
// TODO: avoid serializing twice?
size_t len = root.measureLength();
AsyncWebSocketMessageBuffer* buffer = _ws.makeBuffer(len);
if (buffer) {
root.printTo(reinterpret_cast<char*>(buffer->get()), len + 1);
client->text(buffer);
}
}
void _wsConnected(uint32_t client_id) {
const bool changePassword = (USE_PASSWORD && WEB_FORCE_PASS_CHANGE)
? getAdminPass().equals(ADMIN_PASS)
: false;
if (changePassword) {
StaticJsonBuffer<JSON_OBJECT_SIZE(1)> jsonBuffer;
JsonObject& root = jsonBuffer.createObject();
root["webMode"] = WEB_MODE_PASSWORD;
wsSend(client_id, root);
return;
}
wsPostAll(client_id, _ws_callbacks.on_visible);
wsPostSequence(client_id, _ws_callbacks.on_connected);
wsPostSequence(client_id, _ws_callbacks.on_data);
}
void _wsEvent(AsyncWebSocket * server, AsyncWebSocketClient * client, AwsEventType type, void * arg, uint8_t *data, size_t len){
if (type == WS_EVT_CONNECT) {
client->_tempObject = nullptr;
#ifndef NOWSAUTH
if (!_wsAuth(client)) {
wsSend_P(client->id(), PSTR("{\"message\": 10}"));
DEBUG_MSG_P(PSTR("[WEBSOCKET] Validation check failed\n"));
client->close();
return;
}
#endif
IPAddress ip = client->remoteIP();
DEBUG_MSG_P(PSTR("[WEBSOCKET] #%u connected, ip: %d.%d.%d.%d, url: %s\n"), client->id(), ip[0], ip[1], ip[2], ip[3], server->url());
_wsConnected(client->id());
_wsResetUpdateTimer();
wifiReconnectCheck();
client->_tempObject = new WebSocketIncommingBuffer(_wsParse, true);
} else if(type == WS_EVT_DISCONNECT) {
DEBUG_MSG_P(PSTR("[WEBSOCKET] #%u disconnected\n"), client->id());
if (client->_tempObject) {
delete (WebSocketIncommingBuffer *) client->_tempObject;
}
wifiReconnectCheck();
} else if(type == WS_EVT_ERROR) {
DEBUG_MSG_P(PSTR("[WEBSOCKET] #%u error(%u): %s\n"), client->id(), *((uint16_t*)arg), (char*)data);
} else if(type == WS_EVT_PONG) {
DEBUG_MSG_P(PSTR("[WEBSOCKET] #%u pong(%u): %s\n"), client->id(), len, len ? (char*) data : "");
} else if(type == WS_EVT_DATA) {
//DEBUG_MSG_P(PSTR("[WEBSOCKET] #%u data(%u): %s\n"), client->id(), len, len ? (char*) data : "");
if (!client->_tempObject) return;
WebSocketIncommingBuffer *buffer = (WebSocketIncommingBuffer *)client->_tempObject;
AwsFrameInfo * info = (AwsFrameInfo*)arg;
buffer->data_event(client, info, data, len);
}
}
// TODO: make this generic loop method to queue important ws messages?
// or, if something uses ticker / async ctx to send messages,
// it needs a retry mechanism built into the callback object
void _wsHandleClientData(const bool connected) {
if (!connected && !_ws_client_data.empty()) {
_ws_client_data.pop();
return;
}
if (_ws_client_data.empty()) return;
auto& data = _ws_client_data.front();
// client_id == 0 means we need to send the message to every client
if (data.client_id) {
AsyncWebSocketClient* ws_client = _ws.client(data.client_id);
if (!ws_client) {
_ws_client_data.pop();
return;
}
// wait until we can send the next batch of messages
// XXX: enforce that callbacks send only one message per iteration
if (ws_client->queueIsFull()) {
return;
}
}
// XXX: block allocation will try to create *2 next time,
// likely failing and causing wsSend to reference empty objects
// XXX: arduinojson6 will not do this, but we may need to use per-callback buffers
constexpr const size_t BUFFER_SIZE = 3192;
DynamicJsonBuffer jsonBuffer(BUFFER_SIZE);
JsonObject& root = jsonBuffer.createObject();
data.send(root);
if (data.client_id) {
wsSend(data.client_id, root);
} else {
wsSend(root);
}
yield();
if (data.done()) {
_ws_client_data.pop();
}
}
void _wsLoop() {
const bool connected = wsConnected();
_wsDoUpdate(connected);
_wsHandleClientData(connected);
#if DEBUG_WEB_SUPPORT
_ws_debug.send(connected);
#endif
}
// -----------------------------------------------------------------------------
// Public API
// -----------------------------------------------------------------------------
bool wsConnected() {
return (_ws.count() > 0);
}
bool wsConnected(uint32_t client_id) {
return _ws.hasClient(client_id);
}
ws_callbacks_t& wsRegister() {
return _ws_callbacks;
}
void wsSend(ws_on_send_callback_f callback) {
if (_ws.count() > 0) {
DynamicJsonBuffer jsonBuffer(512);
JsonObject& root = jsonBuffer.createObject();
callback(root);
wsSend(root);
}
}
void wsSend(const char * payload) {
if (_ws.count() > 0) {
_ws.textAll(payload);
}
}
void wsSend_P(PGM_P payload) {
if (_ws.count() > 0) {
char buffer[strlen_P(payload)];
strcpy_P(buffer, payload);
_ws.textAll(buffer);
}
}
void wsSend(uint32_t client_id, ws_on_send_callback_f callback) {
AsyncWebSocketClient* client = _ws.client(client_id);
if (client == nullptr) return;
DynamicJsonBuffer jsonBuffer(512);
JsonObject& root = jsonBuffer.createObject();
callback(root);
wsSend(client_id, root);
}
void wsSend(uint32_t client_id, const char * payload) {
_ws.text(client_id, payload);
}
void wsSend_P(uint32_t client_id, PGM_P payload) {
char buffer[strlen_P(payload)];
strcpy_P(buffer, payload);
_ws.text(client_id, buffer);
}
void wsPost(const ws_on_send_callback_f& cb) {
_ws_client_data.emplace(cb);
}
void wsPost(uint32_t client_id, const ws_on_send_callback_f& cb) {
_ws_client_data.emplace(client_id, cb);
}
void wsPostAll(uint32_t client_id, const ws_on_send_callback_list_t& cbs) {
_ws_client_data.emplace(client_id, cbs, ws_data_t::ALL);
}
void wsPostAll(const ws_on_send_callback_list_t& cbs) {
_ws_client_data.emplace(0, cbs, ws_data_t::ALL);
}
void wsPostSequence(uint32_t client_id, const ws_on_send_callback_list_t& cbs) {
_ws_client_data.emplace(client_id, cbs, ws_data_t::SEQUENCE);
}
void wsPostSequence(uint32_t client_id, ws_on_send_callback_list_t&& cbs) {
_ws_client_data.emplace(client_id, std::forward<ws_on_send_callback_list_t>(cbs), ws_data_t::SEQUENCE);
}
void wsPostSequence(const ws_on_send_callback_list_t& cbs) {
_ws_client_data.emplace(0, cbs, ws_data_t::SEQUENCE);
}
void wsSetup() {
_ws.onEvent(_wsEvent);
webServer()->addHandler(&_ws);
// CORS
const String webDomain = getSetting("webDomain", WEB_REMOTE_DOMAIN);
DefaultHeaders::Instance().addHeader("Access-Control-Allow-Origin", webDomain);
if (!webDomain.equals("*")) {
DefaultHeaders::Instance().addHeader("Access-Control-Allow-Credentials", "true");
}
webServer()->on("/auth", HTTP_GET, _onAuth);
wsRegister()
.onConnected(_wsOnConnected)
.onKeyCheck(_wsOnKeyCheck);
espurnaRegisterLoop(_wsLoop);
}
#endif // WEB_SUPPORT