mhsw-espurna/code/espurna/terminal.cpp

/*

TERMINAL MODULE

Copyright (C) 2016-2019 by Xose Pérez <xose dot perez at gmail dot com>
Copyright (C) 2020 by Maxim Prokhorov <prokhorov dot max at outlook dot com>

*/

#include "espurna.h"

#if TERMINAL_SUPPORT

#include "api.h"
#include "settings.h"
#include "system.h"
#include "telnet.h"
#include "utils.h"
#include "mqtt.h"
#include "wifi.h"
#include "ws.h"

#include "libs/URL.h"
#include "libs/StreamAdapter.h"
#include "libs/PrintString.h"

#include "web_asyncwebprint_impl.h"

#include <algorithm>
#include <vector>
#include <utility>

#include <Schedule.h>
#include <Stream.h>

// not yet CONNECTING or LISTENING
extern struct tcp_pcb *tcp_bound_pcbs;
// accepting or sending data
extern struct tcp_pcb *tcp_active_pcbs;
// // TIME-WAIT status
extern struct tcp_pcb *tcp_tw_pcbs;

namespace {

// Based on libs/StreamInjector.h by Xose Pérez <xose dot perez at gmail dot com> (see git-log for more info)
// Instead of custom write(uint8_t) callback, we provide writer implementation in-place

struct TerminalIO final : public Stream {

    TerminalIO(size_t capacity = 128) :
        _buffer(new char[capacity]),
        _capacity(capacity),
        _write(0),
        _read(0)
    {}

    ~TerminalIO() {
        delete[] _buffer;
    }

    // ---------------------------------------------------------------------
    // Injects data into the internal buffer so we can read() it
    // ---------------------------------------------------------------------

    size_t capacity() {
        return _capacity;
    }

    size_t inject(char ch) {
        _buffer[_write] = ch;
        _write = (_write + 1) % _capacity;
        return 1;
    }

    size_t inject(char *data, size_t len) {
        for (size_t index = 0; index < len; ++index) {
            inject(data[index]);
        }
        return len;
    }

    // ---------------------------------------------------------------------
    // XXX: We are only supporting part of the Print & Stream interfaces
    //      But, we need to be have all pure virtual methods implemented
    // ---------------------------------------------------------------------

    // Return data from the internal buffer
    int available() override {
        unsigned int bytes = 0;
        if (_read > _write) {
            bytes += (_write - _read + _capacity);
        } else if (_read < _write) {
            bytes += (_write - _read);
        }
        return bytes;
    }

    int peek() override {
        int ch = -1;
        if (_read != _write) {
            ch = _buffer[_read];
        }
        return ch;
    }

    int read() override {
        int ch = -1;
        if (_read != _write) {
            ch = _buffer[_read];
            _read = (_read + 1) % _capacity;
        }
        return ch;
    }

    // {Stream,Print}::flush(), see:
    // - https://github.com/esp8266/Arduino/blob/master/cores/esp8266/Print.h
    // - https://github.com/espressif/arduino-esp32/blob/master/cores/esp32/Print.h
    // - https://github.com/arduino/ArduinoCore-API/issues/102
    // Old 2.3.0 expects flush() on Stream, latest puts in in Print
    // We may have to cheat the system and implement everything as Stream to have it available.
    void flush() override {
        // Here, reset reader position so that we return -1 until we have new data
        // writer flushing is implemented below, we don't need it here atm
        _read = _write;
    }

    size_t write(const uint8_t* buffer, size_t size) override {
    // Buffer data until we encounter line break, then flush via Raw debug method
    // (which is supposed to 1-to-1 copy the data, without adding the timestamp)
#if DEBUG_SUPPORT
        if (!size) return 0;
        if (buffer[size-1] == '\0') return 0;
        if (_output.capacity() < (size + 2)) {
            _output.reserve(_output.size() + size + 2);
        }
        _output.insert(_output.end(),
            reinterpret_cast<const char*>(buffer),
            reinterpret_cast<const char*>(buffer) + size
        );
        if (_output.end() != std::find(_output.begin(), _output.end(), '\n')) {
            _output.push_back('\0');
            debugSendRaw(_output.data());
            _output.clear();
        }
#endif
        return size;
    }

    size_t write(uint8_t ch) override {
        uint8_t buffer[1] {ch};
        return write(buffer, 1);
    }

    private:

#if DEBUG_SUPPORT
    std::vector<char> _output;
#endif

    char * _buffer;
    unsigned char _capacity;
    unsigned char _write;
    unsigned char _read;

};

auto _io = TerminalIO(TERMINAL_SHARED_BUFFER_SIZE);
terminal::Terminal _terminal(_io, _io.capacity());

// TODO: re-evaluate how and why this is used
#if SERIAL_RX_ENABLED

constexpr size_t SerialRxBufferSize { 128u };
char _serial_rx_buffer[SerialRxBufferSize];
static unsigned char _serial_rx_pointer = 0;

#endif // SERIAL_RX_ENABLED

// -----------------------------------------------------------------------------
// Commands
// -----------------------------------------------------------------------------

void _terminalHelpCommand(const terminal::CommandContext& ctx) {
    auto names = _terminal.names();

    // XXX: Core's ..._P funcs only allow 2nd pointer to be in PROGMEM,
    //      explicitly load the 1st one
    std::sort(names.begin(), names.end(), [](const __FlashStringHelper* lhs, const __FlashStringHelper* rhs) {
        const String lhs_as_string(lhs);
        return strncasecmp_P(lhs_as_string.c_str(), reinterpret_cast<const char*>(rhs), lhs_as_string.length()) < 0;
    });

    ctx.output.print(F("Available commands:\n"));
    for (auto* name : names) {
        ctx.output.printf("> %s\n", reinterpret_cast<const char*>(name));
    }

    terminalOK(ctx.output);
}

namespace dns {

using Callback = std::function<void(const char* name, const ip_addr_t* addr, void* arg)>;

namespace internal {

struct Task {
    Task() = delete;
    explicit Task(String&& hostname, Callback&& callback) :
        _hostname(std::move(hostname)),
        _callback(std::move(callback))
    {}

    ip_addr_t* addr() {
        return &_addr;
    }

    const char* hostname() const {
        return _hostname.c_str();
    }

    void callback(const char* name, const ip_addr_t* addr, void* arg) {
        _callback(name, addr, arg);
    }

    void callback() {
        _callback(hostname(), addr(), nullptr);
    }

private:
    String _hostname;
    Callback _callback;
    ip_addr_t _addr { IPADDR_NONE };
};

using TaskPtr = std::unique_ptr<Task>;
TaskPtr task;

void callback(const char* name, const ip_addr_t* addr, void* arg) {
    if (task) {
        task->callback(name, addr, arg);
    }
    task.reset();
}

} // namespace internal

bool started() {
    return static_cast<bool>(internal::task);
}

void start(String&& hostname, Callback&& callback) {
    auto task = std::make_unique<internal::Task>(std::move(hostname), std::move(callback));

    switch (dns_gethostbyname(task->hostname(), task->addr(), internal::callback, nullptr)) {
    case ERR_OK:
        task->callback();
        break;
    case ERR_INPROGRESS:
        internal::task = std::move(task);
        break;
    default:
        break;
    }
}

} // namespace dns

void _terminalInitCommands() {

    terminalRegisterCommand(F("COMMANDS"), _terminalHelpCommand);
    terminalRegisterCommand(F("HELP"), _terminalHelpCommand);

    terminalRegisterCommand(F("ERASE.CONFIG"), [](const terminal::CommandContext&) {
        terminalOK();
        customResetReason(CustomResetReason::Terminal);
        eraseSDKConfig();
        *((int*) 0) = 0; // see https://github.com/esp8266/Arduino/issues/1494
    });

    terminalRegisterCommand(F("ADC"), [](const terminal::CommandContext& ctx) {
        const int pin = (ctx.argc == 2)
            ? ctx.argv[1].toInt()
            : A0;

        ctx.output.println(analogRead(pin));
        terminalOK(ctx);
    });

    terminalRegisterCommand(F("GPIO"), [](const terminal::CommandContext& ctx) {
        const int pin = (ctx.argc >= 2)
            ? ctx.argv[1].toInt()
            : -1;

        if ((pin >= 0) && !gpioValid(pin)) {
            terminalError(ctx, F("Invalid pin number"));
            return;
        }

        int start = 0;
        int end = gpioPins();

        switch (ctx.argc) {
        case 3:
            pinMode(pin, OUTPUT);
            digitalWrite(pin, (1 == ctx.argv[2].toInt()));
            break;
        case 2:
            start = pin;
            end = pin + 1;
            // fallthrough into print
        case 1:
            for (auto current = start; current < end; ++current) {
                if (gpioValid(current)) {
                    ctx.output.printf_P(PSTR("%c %s @ GPIO%02d (%s)\n"),
                        gpioLocked(current) ? '*' : ' ',
                        GPEP(current) ? "OUTPUT" : "INPUT ",
                        current,
                        (HIGH == digitalRead(current)) ? "HIGH" : "LOW"
                    );
                }
            }
            break;
        }

        terminalOK(ctx);
    });

    terminalRegisterCommand(F("HEAP"), [](const terminal::CommandContext& ctx) {
        static auto initial = systemInitialFreeHeap();

        auto stats = systemHeapStats();
        ctx.output.printf_P(PSTR("initial: %u, available: %u, fragmentation: %hhu%%\n"),
            initial, stats.available, stats.frag_pct);

        terminalOK(ctx);
    });

    terminalRegisterCommand(F("STACK"), [](const terminal::CommandContext& ctx) {
        ctx.output.printf_P(PSTR("continuation stack initial: %d, free: %u\n"),
            CONT_STACKSIZE, systemFreeStack());
        terminalOK(ctx);
    });

    terminalRegisterCommand(F("INFO"), [](const terminal::CommandContext&) {
        info();
        terminalOK();
    });

    terminalRegisterCommand(F("RESET"), [](const terminal::CommandContext& ctx) {
        if (ctx.argc == 2) {
            auto arg = ctx.argv[1].toInt();
            if (arg < SYSTEM_CHECK_MAX) {
                systemStabilityCounter(arg);
            }
        }

        terminalOK(ctx);
        deferredReset(100, CustomResetReason::Terminal);
    });

    terminalRegisterCommand(F("UPTIME"), [](const terminal::CommandContext& ctx) {
        ctx.output.println(getUptime());
        terminalOK(ctx);
    });

#if SECURE_CLIENT == SECURE_CLIENT_BEARSSL
    terminalRegisterCommand(F("MFLN.PROBE"), [](const terminal::CommandContext& ctx) {
        if (ctx.argc != 3) {
            terminalError(F("[url] [value]"));
            return;
        }

        URL _url(ctx.argv[1]);
        uint16_t requested_mfln = atol(ctx.argv[2].c_str());

        auto client = std::make_unique<BearSSL::WiFiClientSecure>();
        client->setInsecure();

        if (client->probeMaxFragmentLength(_url.host.c_str(), _url.port, requested_mfln)) {
            terminalOK();
        } else {
            terminalError(F("Buffer size not supported"));
        }
    });
#endif

    terminalRegisterCommand(F("HOST"), [](const terminal::CommandContext& ctx) {
        if (ctx.argc != 2) {
            terminalError(ctx, F("HOST <hostname>"));
            return;
        }

        dns::start(String(ctx.argv[1]), [&](const char* name, const ip_addr_t* addr, void*) {
            if (!addr) {
                ctx.output.printf_P(PSTR("%s not found\n"), name);
                return;
            }

            ctx.output.printf_P(PSTR("%s has address %s\n"),
                name, IPAddress(addr).toString().c_str());
        });

        while (dns::started()) {
            delay(100);
        }
    });

    terminalRegisterCommand(F("NETSTAT"), [](const terminal::CommandContext& ctx) {
        auto print = [](Print& out, tcp_pcb* list) {
            for (tcp_pcb* pcb = list; pcb != nullptr; pcb = pcb->next) {
                out.printf_P(PSTR("state %s local %s:%hu remote %s:%hu\n"),
                        tcp_debug_state_str(pcb->state),
                        IPAddress(pcb->local_ip).toString().c_str(),
                        pcb->local_port,
                        IPAddress(pcb->remote_ip).toString().c_str(),
                        pcb->remote_port);
            }
        };

        print(ctx.output, tcp_active_pcbs);
        print(ctx.output, tcp_tw_pcbs);
        print(ctx.output, tcp_bound_pcbs);
    });
}

void _terminalLoop() {

    #if DEBUG_SERIAL_SUPPORT
        while (DEBUG_PORT.available()) {
            _io.inject(DEBUG_PORT.read());
        }
    #endif

    _terminal.process([](terminal::Terminal::Result result) {
        bool out = false;
        switch (result) {
            case terminal::Terminal::Result::CommandNotFound:
                terminalError(terminalDefaultStream(), F("Command not found"));
                out = true;
                break;
            case terminal::Terminal::Result::BufferOverflow:
                terminalError(terminalDefaultStream(), F("Command line buffer overflow"));
                out = true;
                break;
            case terminal::Terminal::Result::Command:
                out = true;
                break;
            case terminal::Terminal::Result::Pending:
                out = false;
                break;
            case terminal::Terminal::Result::Error:
                terminalError(terminalDefaultStream(), F("Unexpected error when parsing command line"));
                out = false;
                break;
            case terminal::Terminal::Result::NoInput:
                out = false;
                break;
        }
        return out;
    });

    #if SERIAL_RX_ENABLED

        while (SERIAL_RX_PORT.available() > 0) {
            char rc = SERIAL_RX_PORT.read();
            _serial_rx_buffer[_serial_rx_pointer++] = rc;
            if ((_serial_rx_pointer == SerialRxBufferSize) || (rc == 10)) {
                terminalInject(_serial_rx_buffer, (size_t) _serial_rx_pointer);
                _serial_rx_pointer = 0;
            }
        }

    #endif // SERIAL_RX_ENABLED

}

#if MQTT_SUPPORT && TERMINAL_MQTT_SUPPORT

void _terminalMqttSetup() {

    mqttRegister([](unsigned int type, const char * topic, const char * payload) {
        if (type == MQTT_CONNECT_EVENT) {
            mqttSubscribe(MQTT_TOPIC_CMD);
            return;
        }

        if (type == MQTT_MESSAGE_EVENT) {
            String t = mqttMagnitude((char *) topic);
            if (!t.startsWith(MQTT_TOPIC_CMD)) return;
            if (!strlen(payload)) return;

            String cmd(payload);
            if (!cmd.endsWith("\r\n") && !cmd.endsWith("\n")) {
                cmd += '\n';
            }

            // TODO: unlike http handler, we have only one output stream
            //       and **must** have a fixed-size output buffer
            //       (wishlist: MQTT client does some magic and we don't buffer twice)
            schedule_function([cmd]() {
                PrintString buffer(TCP_MSS);
                StreamAdapter<const char*> stream(buffer, cmd.c_str(), cmd.c_str() + cmd.length() + 1);

                String out;
                terminal::Terminal handler(stream);
                switch (handler.processLine()) {
                case terminal::Terminal::Result::CommandNotFound:
                    out += F("Command not found");
                    break;
                case terminal::Terminal::Result::Command:
                    out = std::move(buffer);
                default:
                    break;
                }

                if (out.length()) {
                    mqttSendRaw(mqttTopic(MQTT_TOPIC_CMD, false).c_str(), out.c_str(), false);
                }
            });

            return;
        }
    });

}

#endif // MQTT_SUPPORT && TERMINAL_MQTT_SUPPORT

} // namespace

// -----------------------------------------------------------------------------
// Pubic API
// -----------------------------------------------------------------------------

#if TERMINAL_WEB_API_SUPPORT

// XXX: new `apiRegister()` depends that `webServer()` is available, meaning we can't call this setup func
// before the `webSetup()` is called. ATM, just make sure it is in order.

void terminalWebApiSetup() {
#if API_SUPPORT
    apiRegister(getSetting("termWebApiPath", TERMINAL_WEB_API_PATH),
        [](ApiRequest& api) {
            api.handle([](AsyncWebServerRequest* request) {
                AsyncResponseStream *response = request->beginResponseStream("text/plain");
                for (auto* name : _terminal.names()) {
                    response->print(name);
                    response->print("\r\n");
                }

                request->send(response);
            });
            return true;
        },
        [](ApiRequest& api) {
            // TODO: since HTTP spec allows query string to contain repeating keys, allow iteration
            // over every received 'line' to provide a way to call multiple commands at once
            auto cmd = api.param(F("line"));
            if (!cmd.length()) {
                return false;
            }

            if (!cmd.endsWith("\r\n") && !cmd.endsWith("\n")) {
                cmd += '\n';
            }

            api.handle([&](AsyncWebServerRequest* request) {
                AsyncWebPrint::scheduleFromRequest(request, [cmd](Print& print) {
                    StreamAdapter<const char*> stream(print, cmd.c_str(), cmd.c_str() + cmd.length() + 1);
                    terminal::Terminal handler(stream);
                    handler.processLine();
                });
            });

            return true;
        }
    );
#else
    webRequestRegister([](AsyncWebServerRequest* request) {
        String path(F(API_BASE_PATH));
        path += getSetting("termWebApiPath", TERMINAL_WEB_API_PATH);
        if (path != request->url()) {
            return false;
        }

        if (!apiAuthenticate(request)) {
            request->send(403);
            return true;
        }

        auto* cmd_param = request->getParam("line", (request->method() == HTTP_PUT));
        if (!cmd_param) {
            request->send(500);
            return true;
        }

        auto cmd = cmd_param->value();
        if (!cmd.length()) {
            request->send(500);
            return true;
        }

        if (!cmd.endsWith("\r\n") && !cmd.endsWith("\n")) {
            cmd += '\n';
        }

        // TODO: batch requests? processLine() -> process(...)
        AsyncWebPrint::scheduleFromRequest(request, [cmd](Print& print) {
            StreamAdapter<const char*> stream(print, cmd.c_str(), cmd.c_str() + cmd.length() + 1);
            terminal::Terminal handler(stream);
            handler.processLine();
        });

        return true;
    });
#endif // API_SUPPORT
}

#endif // TERMINAL_WEB_API_SUPPORT

Stream & terminalDefaultStream() {
    return (Stream &) _io;
}

size_t terminalCapacity() {
    return _io.capacity();
}

void terminalInject(void *data, size_t len) {
    _io.inject((char *) data, len);
}

void terminalInject(char ch) {
    _io.inject(ch);
}

void terminalRegisterCommand(const __FlashStringHelper* name, terminal::Terminal::CommandFunc func) {
    terminal::Terminal::addCommand(name, func);
};

void terminalOK(Print& print) {
    print.print(F("+OK\n"));
}

void terminalError(Print& print, const String& error) {
    print.printf("-ERROR: %s\n", error.c_str());
}

void terminalOK(const terminal::CommandContext& ctx) {
    terminalOK(ctx.output);
}

void terminalError(const terminal::CommandContext& ctx, const String& error) {
    terminalError(ctx.output, error);
}

void terminalOK() {
    terminalOK(_io);
}

void terminalError(const String& error) {
    terminalError(_io, error);
}

void terminalSetup() {

    // Show DEBUG panel with input
    #if WEB_SUPPORT
        wsRegister()
            .onVisible([](JsonObject& root) { root["cmdVisible"] = 1; });
    #endif

    // Similar to the above, but we allow only very small and in-place outputs.
    #if MQTT_SUPPORT && TERMINAL_MQTT_SUPPORT
        _terminalMqttSetup();
    #endif

    // Initialize default commands
    _terminalInitCommands();

    #if SERIAL_RX_ENABLED
        SERIAL_RX_PORT.begin(SERIAL_RX_BAUDRATE);
    #endif // SERIAL_RX_ENABLED

    // Register loop
    espurnaRegisterLoop(_terminalLoop);

}

#endif // TERMINAL_SUPPORT