signalk_ws_client.h

Go to the documentation of this file.

#ifndef SENSESP_SRC_SENSESP_SIGNALK_SIGNALK_WS_CLIENT_H_
#define SENSESP_SRC_SENSESP_SIGNALK_SIGNALK_WS_CLIENT_H_
 
#include "sensesp.h"
 
#include <algorithm>
 
#include <ArduinoJson.h>
#include <esp_websocket_client.h>
#include <atomic>
#include <functional>
#include <list>
#include <set>
 
#include "sensesp/signalk/signalk_delta_queue.h"
#include "sensesp/system/observablevalue.h"
#include "sensesp/system/task_queue_producer.h"
#include "sensesp/system/valueproducer.h"
#include "sensesp/transforms/integrator.h"
#include "sensesp_base_app.h"
 
// Maximum number of received value/put deltas buffered for processing on the
// main task. Oldest entries are dropped when the buffer is full.
#ifndef SENSESP_MAX_RECEIVED_VALUE_UPDATES
#define SENSESP_MAX_RECEIVED_VALUE_UPDATES 20
#endif
 
// Maximum number of received meta deltas buffered for processing on the main
// task, budgeted independently of value deltas so a metadata burst (~one entry
// per subscribed path at subscribe time when sendMeta=all is enabled) cannot
// evict pending values, and vice versa. Override per board via build_flags if
// you subscribe many paths and need the full burst delivered (more paths =>
// more buffered meta => more RAM).
#ifndef SENSESP_MAX_RECEIVED_META_UPDATES
#define SENSESP_MAX_RECEIVED_META_UPDATES 20
#endif
 
// Signal K WebSocket buffer size in bytes, allocated by esp_websocket_client for
// both the tx and rx buffer. send_delta drops any delta longer than this rather
// than let esp_websocket_client split it across non-blocking writes and abort
// the connection (see signalk_ws_delta_size.h). Override per board via
// build_flags to hold the largest delta a device sends (e.g. a GNSS receiver's
// full-sky satellitesInView); kept small by default for memory-constrained
// boards.
#ifndef SENSESP_SK_WS_BUFFER_SIZE
#define SENSESP_SK_WS_BUFFER_SIZE 1024
#endif
 
namespace sensesp {
 
static const char* NULL_AUTH_TOKEN = "";
 
// HTTP status returned on a WebSocket upgrade when the Signal K server rejects
// the auth token.
static constexpr int kHttpUnauthorized = 401;
 
bool should_clear_token_on_status(int handshake_status);
 
enum class SKWSConnectionState {
  kSKWSDisconnected,
  kSKWSAuthorizing,
  kSKWSConnecting,
  kSKWSConnected,
  // TLS certificate pinning rejected the server certificate. Distinct from a
  // plain disconnect so the UI can tell a cert problem from a network drop.
  kSKWSCertificateError
};
 
class SKWSClient : public FileSystemSaveable,
                   virtual public ValueProducer<SKWSConnectionState> {
 public:
  // main task methods
 
  SKWSClient(const String& config_path,
             std::shared_ptr<SKDeltaQueue> sk_delta_queue,
             const String& server_address, uint16_t server_port,
             bool use_mdns = true);
 
  const String get_server_address() const { return server_address_; }
  uint16_t get_server_port() const { return server_port_; }
 
  virtual bool to_json(JsonObject& root) override final;
  virtual bool from_json(const JsonObject& config) override final;
 
  ValueProducer<int>& get_delta_tx_count_producer() {
    return delta_tx_count_producer_;
  };
 
  ValueProducer<int>& get_delta_rx_count_producer() {
    return delta_rx_count_producer_;
  };
 
  String get_connection_status();
 
  // SK websocket connection methods
 
  void on_disconnected();
  // handshake_status is the HTTP status of a failed WebSocket upgrade (0 if not
  // applicable); a 401 means the auth token was rejected.
  void on_error(int handshake_status);
  void on_connected();
  void on_receive_delta(uint8_t* payload, size_t length);
  void on_receive_updates(JsonDocument& message);
  void on_receive_put(JsonDocument& message);
  void connect();
  void loop();
  bool is_connected();
  void restart();
  bool is_connect_due() const { return millis() >= next_attempt_ms_; }
  void send_delta();
 
  void sendTXT(String& payload);
 
  bool is_ssl_enabled() const { return ssl_enabled_; }
 
  String get_auth_token() const { return auth_token_; }
 
  void set_ssl_enabled(bool enabled) {
    ssl_enabled_ = enabled;
    save();
  }
 
  bool is_send_meta_enabled() const { return send_meta_enabled_; }
  void set_send_meta_enabled(bool enabled) {
    send_meta_enabled_ = enabled;
    save();
  }
 
  bool is_tofu_enabled() const { return tofu_enabled_; }
 
  void set_tofu_enabled(bool enabled) {
    tofu_enabled_ = enabled;
    save();
  }
 
  bool has_tofu_anchor() const {
    return !tofu_ca_pem_.isEmpty() || !tofu_fingerprint_.isEmpty();
  }
 
  bool has_tofu_ca() const { return !tofu_ca_pem_.isEmpty(); }
  const String& get_tofu_ca() const { return tofu_ca_pem_; }
 
  bool has_tofu_fingerprint() const { return !tofu_fingerprint_.isEmpty(); }
  const String& get_tofu_fingerprint() const { return tofu_fingerprint_; }
 
  const String& get_tofu_pin_cn() const { return tofu_pin_cn_; }
  bool is_tofu_pin_ca() const { return tofu_pin_is_ca_; }
 
  const String& get_tofu_san() const { return tofu_san_; }
 
  void reset_tofu() {
    tofu_ca_pem_ = "";
    tofu_fingerprint_ = "";
    tofu_pin_cn_ = "";
    tofu_pin_is_ca_ = false;
    tofu_san_ = "";
    save();
  }
 
  void stash_pending_ca(const String& ca_pem, const String& cn) {
    pending_ca_pem_ = ca_pem;
    pending_fingerprint_ = "";
    pending_cn_ = cn;
    pending_is_ca_ = true;
    pending_valid_ = true;
  }
  void stash_pending_leaf(const String& fingerprint, const String& cn) {
    pending_ca_pem_ = "";
    pending_fingerprint_ = fingerprint;
    pending_cn_ = cn;
    pending_is_ca_ = false;
    pending_valid_ = true;
  }
  void set_pending_san(const String& san) { pending_san_ = san; }
  void clear_pending_tofu() {
    pending_ca_pem_ = "";
    pending_fingerprint_ = "";
    pending_cn_ = "";
    pending_san_ = "";
    pending_is_ca_ = false;
    pending_valid_ = false;
  }
  bool has_pending_ca() const { return pending_valid_ && pending_is_ca_; }
  void commit_pending_tofu() {
    if (!pending_valid_) {
      return;
    }
    tofu_pin_cn_ = pending_cn_;
    tofu_pin_is_ca_ = pending_is_ca_;
    if (pending_is_ca_) {
      tofu_ca_pem_ = pending_ca_pem_;
      tofu_fingerprint_ = "";
      tofu_san_ = pending_san_;  // bind the leaf identity in CA-anchor mode
    } else {
      tofu_fingerprint_ = pending_fingerprint_;
      tofu_ca_pem_ = "";
      tofu_san_ = "";  // leaf mode: the fingerprint binds identity
    }
    clear_pending_tofu();
    save();
  }
 
  void flag_cert_error() { cert_error_.store(true); }
 
  uint32_t client_generation() const { return client_generation_.load(); }
 
 protected:
  // these are the actually used values
  String server_address_ = "";
  uint16_t server_port_ = 80;
  // these are the hardcoded and/or conf file values
  String conf_server_address_ = "";
  uint16_t conf_server_port_ = 0;
  bool use_mdns_ = true;
 
  String client_id_ = "";
  String polling_href_ = "";
  String auth_token_ = NULL_AUTH_TOKEN;
 
  unsigned long next_attempt_ms_ = 0;
  unsigned long connect_interval_ms_ = 2000;
 
  // SSL/TLS configuration
  bool ssl_enabled_ = false;
  bool tofu_enabled_ = true;  // TOFU enabled by default
 
  // Subscribe with ?sendMeta=all so metadata (zones, units, etc) is
  // pushed in-stream rather than requiring REST /meta polls.
  bool send_meta_enabled_ = true;
 
  // TOFU trust anchor. At most one of these is non-empty once captured:
  //   tofu_ca_pem_      — PEM of the pinned issuing CA (CA-anchor mode)
  //   tofu_fingerprint_ — SHA256 hex of the pinned leaf (leaf-fingerprint mode,
  //                       and the legacy on-disk format the migration reads)
  String tofu_fingerprint_ = "";  // SHA256 fingerprint in hex (64 chars)
  String tofu_ca_pem_ = "";       // PEM of pinned CA (CA-anchor mode)
  // TOFU'd leaf identity (normalized DNS SAN set) bound in CA-anchor mode, so a
  // reconnecting leaf must present the same identity AND chain to the pinned CA.
  // This is what makes CA pinning safe against a public CA (e.g. Let's Encrypt):
  // a valid leaf for a different name from the same CA fails the identity check.
  // Empty in leaf-fingerprint mode.
  String tofu_san_ = "";
  // Display-only identity of the pinned cert, captured at pin time (X.509
  // parsing is only available during the handshake). The CN is attacker-
  // controlled at capture and is bounded + sanitized before storage.
  String tofu_pin_cn_ = "";
  bool tofu_pin_is_ca_ = false;
 
  // Candidate anchor seen during the current connection attempt, persisted only
  // after the connection succeeds (commit_pending_tofu) so an unauthenticated
  // MITM handshake cannot plant a trust anchor.
  String pending_ca_pem_ = "";
  String pending_fingerprint_ = "";
  String pending_cn_ = "";
  String pending_san_ = "";
  bool pending_is_ca_ = false;
  bool pending_valid_ = false;
 
  // Set by the verify callback (transport task) on certificate rejection; read
  // by set_connection_state to surface kSKWSCertificateError. Atomic because the
  // callback and state changes may run on different cores.
  std::atomic<bool> cert_error_{false};
 
  TaskQueueProducer<SKWSConnectionState> connection_state_{
      SKWSConnectionState::kSKWSDisconnected, event_loop()};
 
  SKWSConnectionState task_connection_state_ =
      SKWSConnectionState::kSKWSDisconnected;
 
  // Atomic for pointer-atomicity: the connect worker builds/stores it; the
  // SK/event-loop context loads it to send and reaps it via exchange(nullptr)
  // (single check-and-null, so no double-free). Writers are serialized
  // single-owner-at-a-time by auth_job_running_ / teardown_in_progress_.
  // NOTE: the atomic confers pointer-atomicity, not lifetime safety. send_delta
  // and sendTXT run on the event loop, serialized with detach_teardown (same
  // context), so they cannot hold a handle across its reap. Transport-task
  // senders (on_connected -> subscribe_listeners, on_receive_put's response)
  // are made safe by esp_websocket_client_stop() joining the transport task
  // before destroy() frees the struct. See #1033.
  std::atomic<esp_websocket_client_handle_t> client_{nullptr};
  // Bumped on every teardown so the (singleton) event handler can drop late
  // callbacks from a client that has been handed off for destruction. The
  // handler is registered with the generation current at build time; an event
  // whose generation != client_generation_ comes from a torn-down client.
  std::atomic<uint32_t> client_generation_{0};
  // True while a detached task is stopping+destroying a previous client.
  // Bring-up is deferred until it clears, so at most one client ever exists.
  std::atomic<bool> teardown_in_progress_{false};
  // True while a one-shot worker is running a (blocking) connect attempt
  // (mDNS / SSL-detect / access-request / poll legs). The dispatcher skips
  // while it is set, so at most one attempt runs at a time.
  std::atomic<bool> auth_job_running_{false};
  // Holds a handle whose detached reaper task failed to spawn (OOM); the event
  // loop retries the spawn each tick rather than reaping it synchronously (which
  // would block the loop). At most one at a time (teardown_in_progress_ defers
  // bring-up until it is reaped).
  std::atomic<esp_websocket_client_handle_t> pending_teardown_{nullptr};
  std::shared_ptr<SKDeltaQueue> sk_delta_queue_;
  TaskQueueProducer<int> delta_tx_tick_producer_{0, event_loop(), 990};
  Integrator<int, int> delta_tx_count_producer_{1, 0, ""};
  Integrator<int, int> delta_rx_count_producer_{1, 0, ""};
 
  uint32_t last_oversize_log_ms_ = 0;
 
  struct ReceivedUpdate {
    bool is_meta = false;
    JsonDocument doc;
  };
 
  StaticSemaphore_t received_updates_semaphore_buffer_;
  SemaphoreHandle_t received_updates_semaphore_ =
      xSemaphoreCreateRecursiveMutexStatic(&received_updates_semaphore_buffer_);
  std::list<ReceivedUpdate> received_updates_{};
 
  // methods for all tasks
 
  bool take_received_updates_semaphore(unsigned long int timeout_ms = 0) {
    if (timeout_ms == 0) {
      return xSemaphoreTakeRecursive(received_updates_semaphore_,
                                     portMAX_DELAY) == pdTRUE;
    } else {
      return xSemaphoreTakeRecursive(received_updates_semaphore_,
                                     timeout_ms) == pdTRUE;
    }
  }
  void release_received_updates_semaphore() {
    xSemaphoreGiveRecursive(received_updates_semaphore_);
  }
 
  void enqueue_received_update(ReceivedUpdate&& update);
 
  // main task methods
 
  void process_received_updates();
 
  // SK websocket connection methods
 
#ifndef SENSESP_SSL_SUPPORT
  // Validate the auth token over plain HTTP before opening the WebSocket
  // stream. Only used on non-SSL builds; SSL builds validate it on the upgrade
  // itself (see connect()) to avoid a second TLS handshake that fragments the
  // heap.
  void test_token(const String host, const uint16_t port);
#endif
  void send_access_request(const String host, const uint16_t port);
  void poll_access_request(const String host, const uint16_t port,
                           const String href);
  void connect_ws(const String& host, const uint16_t port);
  void detach_teardown();
  void reap_async(esp_websocket_client_handle_t old);
  static void teardown_task(void* arg);
  void run_connect_attempt();
  static void connect_worker(void* arg);
  void subscribe_listeners();
  bool get_mdns_service(String& server_address, uint16_t& server_port);
  bool detect_ssl();
 
  void set_connection_state(SKWSConnectionState state) {
    // A certificate rejection during this attempt (flagged by the verify
    // callback) is surfaced distinctly instead of as a generic disconnect.
    // This is the single chokepoint, so it covers all four TLS paths — the
    // three esp_http_client requests and the websocket — without touching each
    // call site. A fresh attempt (authorizing/connecting) or success clears the
    // flag.
    if (state == SKWSConnectionState::kSKWSDisconnected && cert_error_.load()) {
      state = SKWSConnectionState::kSKWSCertificateError;
    } else if (state != SKWSConnectionState::kSKWSDisconnected &&
               state != SKWSConnectionState::kSKWSCertificateError) {
      cert_error_.store(false);
    }
    task_connection_state_ = state;
    connection_state_.set(state);
  }
  SKWSConnectionState get_connection_state() { return task_connection_state_; }
 
  void schedule_reconnect() {
    next_attempt_ms_ = millis() + connect_interval_ms_ +
                       (esp_random() % (connect_interval_ms_ / 4 + 1));
    connect_interval_ms_ =
        std::min(connect_interval_ms_ * 2, (unsigned long)60000);
  }
 
  void reset_reconnect_interval() {
    connect_interval_ms_ = 2000;
  }
};
 
inline const String ConfigSchema(const SKWSClient& obj) {
  return "{\"type\":\"object\",\"properties\":{"
         "\"ssl_enabled\":{\"title\":\"SSL/TLS Enabled\",\"type\":\"boolean\"},"
         "\"tofu_enabled\":{\"title\":\"TOFU Verification\",\"type\":\"boolean\"},"
         "\"tofu_pin_cn\":{\"title\":\"Pinned Certificate\",\"type\":\"string\",\"readOnly\":true},"
         "\"tofu_pin_is_ca\":{\"title\":\"Pinned as CA\",\"type\":\"boolean\",\"readOnly\":true},"
         "\"send_meta_enabled\":{\"title\":\"Subscribe with sendMeta=all\","
         "\"description\":\"Request metadata deltas (units, zones, displayName, displayUnits) over the WS stream. Disable only for constrained clients that ignore them.\","
         "\"type\":\"boolean\"}"
         "}}";
}
 
inline bool ConfigRequiresRestart(const SKWSClient& obj) { return true; }
 
}  // namespace sensesp
 
#endif