// SPDX-License-Identifier: GPL-3.0-or-later // Copyright (C) 2025 - 2026, Patrick Schurig / ETM PowerSync #include "optimizerclient.h" #include "optimizersecret.h" #include #include #include #include #include #include #include #include #include #include #include #include #include Q_DECLARE_LOGGING_CATEGORY(dcNymeaEnergy) namespace { // Plus long token lu ou écrit. Un secret partagé n'a aucune raison d'approcher cette taille ; la // borne évite qu'un fichier posé par erreur (une archive, un journal) parte au handshake. constexpr int kMaxTokenBytes = 4096; // Majeure d'une version semver « M.m » : ce qui précède le premier point. QString majorOf(const QString &version) { return version.section(QLatin1Char('.'), 0, 0); } } // namespace // ─── OptimizerEndpoint ────────────────────────────────────────────────────────────────────── OptimizerEndpoint OptimizerEndpoint::parse(const QString &text) { OptimizerEndpoint e; static const QString unixScheme = QStringLiteral("unix://"); static const QString tcpScheme = QStringLiteral("tcp://"); if (text.startsWith(unixScheme)) { const QString path = text.mid(unixScheme.size()); if (path.size() < 2 || !path.startsWith(QLatin1Char('/'))) return e; e.kind = Unix; e.path = path; return e; } if (!text.startsWith(tcpScheme)) return e; const QString rest = text.mid(tcpScheme.size()); QString host; QString portText; if (rest.startsWith(QLatin1Char('['))) { const int close = rest.indexOf(QLatin1Char(']')); if (close < 0 || rest.mid(close + 1, 1) != QLatin1String(":")) return e; host = rest.mid(1, close - 1); portText = rest.mid(close + 2); } else { const int colon = rest.indexOf(QLatin1Char(':')); if (colon < 0 || rest.indexOf(QLatin1Char(':'), colon + 1) >= 0) return e; // IPv6 sans crochets : ambigu, refusé host = rest.left(colon); portText = rest.mid(colon + 1); } bool ok = false; const uint port = portText.toUInt(&ok); // toUInt accepte « +7400 » et les espaces de tête : on exige des chiffres seuls. const bool digitsOnly = !portText.isEmpty() && std::all_of(portText.cbegin(), portText.cend(), [](QChar c) { return c.isDigit(); }); if (host.isEmpty() || !ok || !digitsOnly || port == 0 || port > 65535) return e; e.kind = Tcp; e.host = host; e.port = static_cast(port); return e; } QString OptimizerEndpoint::toString() const { switch (kind) { case Unix: return QStringLiteral("unix://") + path; case Tcp: if (host.contains(QLatin1Char(':'))) return QStringLiteral("tcp://[%1]:%2").arg(host).arg(port); return QStringLiteral("tcp://%1:%2").arg(host).arg(port); case Invalid: break; } return QStringLiteral("(adresse invalide)"); } // ─── OptimizerClientClock ─────────────────────────────────────────────────────────────────── OptimizerClientClock::OptimizerClientClock(QObject *parent) : QObject(parent) , m_timer(new QTimer(this)) { m_elapsed.start(); m_timer->setSingleShot(true); connect(m_timer, &QTimer::timeout, this, &OptimizerClientClock::wakeUp); } qint64 OptimizerClientClock::nowMs() const { return m_elapsed.elapsed(); } void OptimizerClientClock::armWakeUp(qint64 delayMs) { m_timer->start(static_cast(qBound(qint64(0), delayMs, qint64(std::numeric_limits::max())))); } void OptimizerClientClock::disarmWakeUp() { m_timer->stop(); } // ─── OptimizerClient ──────────────────────────────────────────────────────────────────────── OptimizerClient::OptimizerClient(const OptimizerEndpoint &endpoint, const QString &tokenPath, const QString &engineVersion, OptimizerClientClock *clock, QObject *parent) : QObject(parent) , m_endpoint(endpoint) , m_tokenPath(tokenPath) , m_engineVersion(engineVersion) , m_clock(clock ? clock : new OptimizerClientClock(this)) { connect(m_clock, &OptimizerClientClock::wakeUp, this, &OptimizerClient::onWakeUp); } OptimizerClient::~OptimizerClient() { // Pas de stop() : il émettrait stateChanged() vers des récepteurs peut-être déjà détruits. disconnect(m_clock, nullptr, this, nullptr); m_clock->disarmWakeUp(); dropSocket(); } QString OptimizerClient::defaultTokenPath() { return NymeaSettings::settingsPath() + QLatin1Char('/') + QLatin1String(kTokenFileName); } bool OptimizerClient::writeToken(const QString &path, const QByteArray &token, QString *error) { auto fail = [error](const QString &why) { if (error) *error = why; return false; }; if (token.isEmpty()) { if (QFile::exists(path) && !QFile::remove(path)) return fail(QStringLiteral("retrait du fichier de token impossible (%1)").arg(path)); return true; } // Le token est relu avec trimmed() : un blanc en bord ou un caractère de contrôle ne // survivrait pas à l'aller-retour, et le handshake échouerait sans raison visible. if (token.size() > kMaxTokenBytes) return fail(QStringLiteral("token refusé : plus de %1 octets").arg(kMaxTokenBytes)); for (const char c : token) { if (static_cast(c) <= 0x20 || c == 0x7f) return fail(QStringLiteral("token refusé : blanc ou caractère de contrôle")); } const QFileDevice::Permissions ownerOnly = QFileDevice::ReadOwner | QFileDevice::WriteOwner; QSaveFile file(path); if (!file.open(QIODevice::WriteOnly)) return fail(QStringLiteral("ouverture de %1 impossible : %2").arg(path, file.errorString())); // Droits posés sur le temporaire AVANT d'y écrire le secret (fchmod sur le descripteur) : // il n'existe aucun instant où le token est sur disque avec les droits de l'umask. Le // renommage du commit les conserve. C'est aussi ce qui resserre un fichier PRÉEXISTANT : // QSaveFile recopie ses droits sur le temporaire à l'ouverture, ils sont écrasés ici. if (!file.setPermissions(ownerOnly)) { file.cancelWriting(); return fail(QStringLiteral("droits 0600 impossibles sur %1 : %2").arg(path, file.errorString())); } if (file.write(token) != token.size()) { file.cancelWriting(); return fail(QStringLiteral("écriture de %1 impossible : %2").arg(path, file.errorString())); } if (!file.commit()) return fail(QStringLiteral("validation de %1 impossible : %2").arg(path, file.errorString())); return true; } void OptimizerClient::start() { if (m_state != Stopped) return; if (!m_endpoint.isValid()) { m_lastErrorCode = QStringLiteral("invalid-endpoint"); qCWarning(dcNymeaEnergy()) << "[Optimiseur] adresse invalide — client non démarré, " "stratégie de repli seule"; return; } qCInfo(dcNymeaEnergy()).noquote() << "[Optimiseur] démarrage du client vers" << m_endpoint.toString(); connectNow(); } void OptimizerClient::stop() { if (m_state == Stopped) return; m_pendingId = 0; m_handshakeDeadline = -1; m_requestDeadline = -1; m_reconnectAt = -1; dropSocket(); m_clock->disarmWakeUp(); qCInfo(dcNymeaEnergy()).noquote() << "[Optimiseur] client arrêté (" << m_endpoint.toString() << ")"; setState(Stopped); } quint64 OptimizerClient::requestPlan(const QVariantMap ¶ms) { if (m_state != Ready) return 0; if (m_pendingId != 0) qCDebug(dcNymeaEnergy()) << "[Optimiseur] requête" << m_pendingId << "remplacée, sa réponse sera ignorée"; const quint64 id = ++m_lastId; m_pendingId = id; m_requestDeadline = m_clock->nowMs() + kRequestTimeoutMs; send(QVariantMap{ {QStringLiteral("jsonrpc"), QStringLiteral("2.0")}, {QStringLiteral("id"), id}, {QStringLiteral("method"), QStringLiteral("GetPlan")}, {QStringLiteral("params"), params}, }); // Un échec d'écriture n'est pas traité ici : le socket signale sa perte (→ planFailed // « disconnected »), et à défaut les 10 s bornent l'attente. Émettre planFailed() avant // d'avoir rendu l'identifiant le rendrait inattribuable par l'appelant. rearm(); return id; } void OptimizerClient::setState(State state) { if (m_state == state) return; m_state = state; emit stateChanged(state); } void OptimizerClient::connectNow() { dropSocket(); m_reconnectAt = -1; m_buffer.clear(); m_skippingLine = false; m_errorsLoggedThisConnection.clear(); m_tokenSent = false; m_handshakeDeadline = m_clock->nowMs() + kHandshakeTimeoutMs; setState(Connecting); rearm(); if (m_endpoint.kind == OptimizerEndpoint::Unix) { auto *socket = new QLocalSocket(this); m_socket = socket; connect(socket, &QLocalSocket::connected, this, &OptimizerClient::onConnected); connect(socket, &QLocalSocket::disconnected, this, &OptimizerClient::onSocketLost); connect(socket, &QLocalSocket::errorOccurred, this, &OptimizerClient::onSocketLost); connect(socket, &QLocalSocket::readyRead, this, &OptimizerClient::onReadyRead); // Peut échouer DANS l'appel (chemin absent) : onSocketLost() s'exécute alors avant le // retour et aura déjà planifié la tentative suivante. Rien ne doit suivre cette ligne. socket->connectToServer(m_endpoint.path); } else { auto *socket = new QTcpSocket(this); m_socket = socket; connect(socket, &QTcpSocket::connected, this, &OptimizerClient::onConnected); connect(socket, &QTcpSocket::disconnected, this, &OptimizerClient::onSocketLost); connect(socket, &QTcpSocket::errorOccurred, this, &OptimizerClient::onSocketLost); connect(socket, &QTcpSocket::readyRead, this, &OptimizerClient::onReadyRead); socket->connectToHost(m_endpoint.host, m_endpoint.port); } } void OptimizerClient::onConnected() { if (sender() != m_socket || m_state != Connecting) return; setState(Handshaking); QVariantMap params{ {QStringLiteral("protocolVersion"), QLatin1String(kProtocolVersion)}, {QStringLiteral("engine"), QStringLiteral("etm-powersync-energy-plugin-etm/") + m_engineVersion}, }; // Relu à CHAQUE connexion (décision Q2) : un token remplacé vaut dès la tentative suivante. const QByteArray token = readToken(); m_tokenSent = !token.isEmpty(); m_tokenConnexion = QString::fromUtf8(token); if (m_tokenSent) params.insert(QStringLiteral("token"), QString::fromUtf8(token)); m_handshakeId = ++m_lastId; // Ce message porte le secret : il n'est JAMAIS journalisé, à aucun niveau. send(QVariantMap{ {QStringLiteral("jsonrpc"), QStringLiteral("2.0")}, {QStringLiteral("id"), m_handshakeId}, {QStringLiteral("method"), QStringLiteral("Handshake")}, {QStringLiteral("params"), params}, }); qCDebug(dcNymeaEnergy()) << "[Optimiseur] connecté, handshake envoyé ( id" << m_handshakeId << ", token" << (m_tokenSent ? "configuré" : "absent") << ")"; } void OptimizerClient::onSocketLost() { if (sender() != m_socket) return; QString detail = m_socket ? m_socket->errorString() : QString(); switch (m_state) { case Connecting: failConnection(QStringLiteral("connect-failed"), detail); break; case Handshaking: if (m_tokenSent) detail += QStringLiteral(" — un token a été envoyé : l'optimiseur attend-il celui-là ?"); failConnection(QStringLiteral("closed-during-handshake"), detail); break; case Ready: failConnection(QStringLiteral("connection-lost"), detail); break; case Stopped: case Waiting: break; } } void OptimizerClient::onReadyRead() { QIODevice *socket = m_socket; if (sender() != socket || !socket) return; m_buffer.append(socket->readAll()); while (m_socket == socket) { const int newline = m_buffer.indexOf('\n'); if (m_skippingLine) { // Reste d'une ligne déjà déclarée trop longue : jeté sans être analysé. if (newline < 0) { m_buffer.clear(); return; } m_buffer.remove(0, newline + 1); m_skippingLine = false; continue; } if (newline < 0) { if (m_buffer.size() > kMaxLineBytes) { countProtocolError(QStringLiteral("line-too-long"), QStringLiteral("plus de %1 octets sans fin de ligne").arg(kMaxLineBytes)); m_buffer.clear(); m_skippingLine = true; } return; } const QByteArray line = m_buffer.left(newline); m_buffer.remove(0, newline + 1); if (line.size() > kMaxLineBytes) { countProtocolError(QStringLiteral("line-too-long"), QStringLiteral("%1 octets").arg(line.size())); continue; } handleLine(line); // peut fermer la connexion : la garde de boucle le voit } } void OptimizerClient::handleLine(const QByteArray &line) { if (line.trimmed().isEmpty()) return; QJsonParseError parseError; const QJsonDocument doc = QJsonDocument::fromJson(line, &parseError); if (parseError.error != QJsonParseError::NoError || !doc.isObject()) { // Le contenu n'est pas journalisé : il peut être volumineux, et il vient d'ailleurs. countProtocolError(QStringLiteral("invalid-json"), QStringLiteral("%1 octets").arg(line.size())); return; } // Correctif A (brief 4 T7) : un optimiseur peut RECOPIER le token qu'il a reçu. Masqué ICI, // avant toute lecture : nom, version, méthode, plan, motifs — rien en aval ne voit le secret, // ni le journal, ni les traces, ni l'app. int echos = 0; const QJsonObject message = OptimizerSecret::masquerJson(doc.object(), m_tokenConnexion, &echos); if (echos > 0 && !m_echoTokenSignale) { m_echoTokenSignale = true; qCWarning(dcNymeaEnergy()) << "[Optimiseur] l'optimiseur recopie le token dans ses réponses — masqué" " à la réception (journal, traces, app) ; signalé une fois"; } const QJsonValue idValue = message.value(QStringLiteral("id")); if (idValue.isUndefined() || idValue.isNull()) { const QJsonValue method = message.value(QStringLiteral("method")); if (method.isString()) { // Notification JSON-RPC (§8 UpdateLearned) : reçue, annoncée, rien de plus ici. qCDebug(dcNymeaEnergy()) << "[Optimiseur] notification" << method.toString() << "reçue"; emit notificationReceived(method.toString(), message.value(QStringLiteral("params")).toObject().toVariantMap()); return; } countProtocolError(QStringLiteral("missing-id"), QStringLiteral("message sans identifiant")); return; } const double raw = idValue.toDouble(-1); if (!idValue.isDouble() || raw < 1 || raw > 9007199254740992.0 || std::floor(raw) != raw) { countProtocolError(QStringLiteral("unknown-id"), QStringLiteral("identifiant non entier")); return; } const quint64 id = static_cast(raw); if (m_state == Handshaking && id == m_handshakeId) { handleHandshakeReply(message); } else if (m_pendingId != 0 && id == m_pendingId) { handlePlanReply(id, message); } else if (id <= m_lastId) { // Requête remplacée ou expirée : sa réponse arrive trop tard, elle ne commande rien. ++m_staleResponses; qCDebug(dcNymeaEnergy()) << "[Optimiseur] réponse obsolète ignorée ( id" << id << ")"; } else { countProtocolError(QStringLiteral("unknown-id"), QStringLiteral("identifiant %1 jamais émis").arg(id)); } } void OptimizerClient::handleHandshakeReply(const QJsonObject &message) { if (message.contains(QStringLiteral("error"))) { // Seul le CODE est journalisé : un serveur peut recopier n'importe quoi dans son // message, y compris le token qu'il vient de refuser. const int code = message.value(QStringLiteral("error")).toObject().value(QStringLiteral("code")).toInt(); QString detail = QStringLiteral("erreur JSON-RPC %1").arg(code); if (m_tokenSent) detail += QStringLiteral(" — un token a été envoyé : l'optimiseur attend-il celui-là ?"); failConnection(QStringLiteral("handshake-refused"), detail); return; } const QJsonValue resultValue = message.value(QStringLiteral("result")); if (!resultValue.isObject()) { failConnection(QStringLiteral("invalid-handshake"), QStringLiteral("result absent ou non objet")); return; } const QJsonObject result = resultValue.toObject(); const QJsonValue version = result.value(QStringLiteral("protocolVersion")); if (!version.isString() || version.toString().isEmpty()) { failConnection(QStringLiteral("invalid-handshake"), QStringLiteral("protocolVersion absent")); return; } if (majorOf(version.toString()) != majorOf(QLatin1String(kProtocolVersion))) { failConnection(QStringLiteral("major-mismatch"), QStringLiteral("optimiseur en %1, moteur en %2 — majeures différentes, refusé (§3)") .arg(version.toString().left(32), QLatin1String(kProtocolVersion))); return; } const QJsonValue capabilities = result.value(QStringLiteral("capabilities")); if (!capabilities.isUndefined() && !capabilities.isObject()) { // §3 : présent, il RESTREINT les kinds acceptés. Illisible, on ne peut pas l'appliquer. failConnection(QStringLiteral("invalid-handshake"), QStringLiteral("capabilities n'est pas un objet")); return; } m_hasCapabilities = capabilities.isObject(); m_capabilities = capabilities.toObject().toVariantMap(); m_serverName = result.value(QStringLiteral("name")).toString(); m_serverProtocolVersion = version.toString(); m_serverPlanning = result.value(QStringLiteral("planning")).toBool(); m_handshakeDeadline = -1; m_backoffMs = kBackoffInitialMs; const int failuresBefore = m_failureStreak; m_failureStreak = 0; m_failureStreakCode.clear(); setState(Ready); rearm(); qCInfo(dcNymeaEnergy()).noquote() << QStringLiteral("[Optimiseur] connecté à %1 : « %2 », protocole %3, capabilities %4%5") .arg(m_endpoint.toString(), m_serverName.left(64), m_serverProtocolVersion.left(32), m_hasCapabilities ? QStringLiteral("annoncées") : QStringLiteral("absentes"), failuresBefore > 0 ? QStringLiteral(" (après %1 échec(s))").arg(failuresBefore) : QString()); emit ready(); } void OptimizerClient::handlePlanReply(quint64 id, const QJsonObject &message) { m_pendingId = 0; m_requestDeadline = -1; rearm(); if (message.contains(QStringLiteral("error"))) { const int code = message.value(QStringLiteral("error")).toObject().value(QStringLiteral("code")).toInt(); m_lastErrorCode = QStringLiteral("rpc-error"); qCWarning(dcNymeaEnergy()) << "[Optimiseur] GetPlan" << id << ": erreur JSON-RPC" << code; emit planFailed(id, QStringLiteral("rpc-error")); return; } const QJsonValue result = message.value(QStringLiteral("result")); if (!result.isObject()) { m_lastErrorCode = QStringLiteral("invalid-response"); qCWarning(dcNymeaEnergy()) << "[Optimiseur] GetPlan" << id << ": réponse sans result objet"; emit planFailed(id, QStringLiteral("invalid-response")); return; } qCDebug(dcNymeaEnergy()) << "[Optimiseur] GetPlan" << id << ": réponse reçue"; emit planReceived(id, result.toObject().toVariantMap()); } void OptimizerClient::failConnection(const QString &code, const QString &detail) { const quint64 pending = m_pendingId; m_pendingId = 0; m_requestDeadline = -1; m_handshakeDeadline = -1; dropSocket(); m_lastErrorCode = code; const qint64 delay = m_backoffMs; // Optimiseur absent : une tentative par minute au plafond. La PREMIÈRE de chaque série est // annoncée ; les suivantes de même nature sont comptées, et la série se clôt en qCInfo au // retour (« après N échec(s) »). ++m_failureStreak; if (code != m_failureStreakCode) { m_failureStreakCode = code; qCWarning(dcNymeaEnergy()).noquote() << QStringLiteral("[Optimiseur] %1 (%2) : %3 — nouvelle tentative dans %4 s, repli rule-based") .arg(m_endpoint.toString(), code, detail).arg(delay / 1000.0); } else { qCDebug(dcNymeaEnergy()).noquote() << QStringLiteral("[Optimiseur] %1 (%2), échec n°%3 — nouvelle tentative dans %4 s") .arg(m_endpoint.toString(), code).arg(m_failureStreak).arg(delay / 1000.0); } scheduleReconnect(); if (pending != 0) emit planFailed(pending, QStringLiteral("disconnected")); } void OptimizerClient::dropSocket() { QIODevice *socket = m_socket; if (!socket) return; m_socket = nullptr; socket->disconnect(this); if (auto *local = qobject_cast(socket)) local->abort(); else if (auto *tcp = qobject_cast(socket)) tcp->abort(); socket->deleteLater(); m_buffer.clear(); m_tokenConnexion.clear(); m_skippingLine = false; } void OptimizerClient::scheduleReconnect() { m_reconnectAt = m_clock->nowMs() + m_backoffMs; m_backoffMs = qMin(m_backoffMs * 2, kBackoffMaxMs); setState(Waiting); rearm(); } void OptimizerClient::onWakeUp() { const qint64 now = m_clock->nowMs(); if (m_handshakeDeadline >= 0 && now >= m_handshakeDeadline && (m_state == Connecting || m_state == Handshaking)) { failConnection(QStringLiteral("handshake-timeout"), QStringLiteral("pas de handshake valide en %1 s").arg(kHandshakeTimeoutMs / 1000)); } if (m_requestDeadline >= 0 && now >= m_requestDeadline && m_pendingId != 0) { const quint64 id = m_pendingId; m_pendingId = 0; m_requestDeadline = -1; m_lastErrorCode = QStringLiteral("timeout"); // La connexion est conservée : une réponse tardive sera reconnue obsolète par son id. qCWarning(dcNymeaEnergy()) << "[Optimiseur] GetPlan" << id << ": pas de réponse en" << kRequestTimeoutMs / 1000 << "s"; emit planFailed(id, QStringLiteral("timeout")); } if (m_reconnectAt >= 0 && now >= m_reconnectAt && m_state == Waiting) connectNow(); rearm(); } void OptimizerClient::rearm() { qint64 next = -1; for (const qint64 deadline : {m_handshakeDeadline, m_requestDeadline, m_reconnectAt}) { if (deadline >= 0 && (next < 0 || deadline < next)) next = deadline; } if (next < 0) m_clock->disarmWakeUp(); else m_clock->armWakeUp(qMax(0, next - m_clock->nowMs())); } void OptimizerClient::countProtocolError(const QString &code, const QString &detail) { ++m_protocolErrors; m_lastErrorCode = code; // Un serveur fautif peut en envoyer à chaque ligne : la première de chaque nature est // journalisée par connexion, le compteur porte les autres. if (!m_errorsLoggedThisConnection.contains(code)) { m_errorsLoggedThisConnection.append(code); qCWarning(dcNymeaEnergy()).noquote() << QStringLiteral("[Optimiseur] ligne écartée (%1) : %2 — connexion conservée ; " "les suivantes de même nature sont comptées sans être journalisées") .arg(code, detail); } else { qCDebug(dcNymeaEnergy()).noquote() << QStringLiteral("[Optimiseur] ligne écartée (%1)").arg(code); } emit protocolError(code); } bool OptimizerClient::send(const QVariantMap &message) { if (!m_socket) return false; QByteArray line = QJsonDocument(QJsonObject::fromVariantMap(message)).toJson(QJsonDocument::Compact); line.append('\n'); return m_socket->write(line) == line.size(); } QByteArray OptimizerClient::readToken() { if (m_tokenPath.isEmpty()) return {}; QFile file(m_tokenPath); if (!file.exists()) return {}; if (!file.open(QIODevice::ReadOnly)) { qCWarning(dcNymeaEnergy()).noquote() << QStringLiteral("[Optimiseur] fichier de token %1 illisible (%2) — handshake sans token") .arg(m_tokenPath, file.errorString()); return {}; } const QFileDevice::Permissions tooWide = QFileDevice::ReadGroup | QFileDevice::WriteGroup | QFileDevice::ExeGroup | QFileDevice::ReadOther | QFileDevice::WriteOther | QFileDevice::ExeOther; if (file.permissions() & tooWide) { qCWarning(dcNymeaEnergy()).noquote() << QStringLiteral("[Optimiseur] fichier de token %1 lisible au-delà de son propriétaire " "(attendu 0600) — utilisé quand même").arg(m_tokenPath); } return file.read(kMaxTokenBytes).trimmed(); }