feat: add OpenAI-compatible AI provider plugin with SSE streaming support

- Implemented the OpenAI-compatible AI provider plugin, including configuration, chat, and chat_stream functionalities. - Added support for SSE streaming and tool calls. - Integrated Boost.JSON for JSON handling. - Created CMake configuration for the plugin. - Added error handling and logging throughout the plugin.
2026-05-31 05:37:04 +08:00
parent f6cb51b40a
commit ba7382db2a
61 changed files with 163 additions and 147 deletions
--- a/plugins_middle/network/CMakeLists.txt
+++ b/plugins_middle/network/CMakeLists.txt
@@ -0,0 +1,16 @@
+find_package(Boost REQUIRED CONFIG)
+find_package(OpenSSL REQUIRED CONFIG)
+
+add_library(plugin-network SHARED src/network_plugin.cpp)
+
+target_link_libraries(plugin-network PRIVATE
+    dstalk
+    boost::boost
+    openssl::openssl
+)
+
+set_target_properties(plugin-network PROPERTIES
+    PREFIX ""
+    LIBRARY_OUTPUT_DIRECTORY "${CMAKE_BINARY_DIR}/plugins"
+    RUNTIME_OUTPUT_DIRECTORY "${CMAKE_BINARY_DIR}/plugins"
+)
--- a/plugins_middle/network/src/network_plugin.cpp
+++ b/plugins_middle/network/src/network_plugin.cpp
@@ -0,0 +1,394 @@
+/*
+ * @file network_plugin.cpp
+ * @brief Network plugin: HTTP/HTTPS POST and streaming via Boost.Beast + OpenSSL.
+ * 网络插件：基于 Boost.Beast + OpenSSL 的 HTTP/HTTPS POST 和流式传输。
+ * Copyright (c) 2026 dstalk contributors. GPLv3.
+ */
+
+// MSVC 14.16 (VS 2017) 不提供 std::to_address (C++20) / MSVC 14.16 (VS 2017) doesn't provide std::to_address (C++20)
+#define BOOST_ASIO_DISABLE_STD_TO_ADDRESS
+
+#include "dstalk/dstalk_host.h"
+#include "dstalk/dstalk_services.h"
+
+#include <boost/asio/connect.hpp>
+#include <boost/asio/ip/tcp.hpp>
+#include <boost/asio/ssl.hpp>
+#include <boost/asio/steady_timer.hpp>
+#include <boost/beast/core.hpp>
+#include <boost/beast/http.hpp>
+#include <boost/beast/ssl.hpp>
+#include <boost/beast/version.hpp>
+
+#include <chrono>
+#include <cstdio>
+#include <cstdlib>
+#include <cstring>
+#include <functional>
+#include <string>
+#include <string_view>
+#include <unordered_map>
+
+namespace beast = boost::beast;
+namespace http  = beast::http;
+namespace asio  = boost::asio;
+namespace ssl   = boost::asio::ssl;
+using tcp = asio::ip::tcp;
+
+// ============================================================
+// 全局状态 / Global state
+// ============================================================
+static const dstalk_host_api_t* g_host = nullptr;
+static dstalk_config_service_t* g_config_svc = nullptr;
+
+// ============================================================
+// 极简 JSON 头解析器 / Minimal JSON header parser
+// 将 {"key1":"value1","key2":"value2"} 解析到 unordered_map / Parses {"key1":"value1","key2":"value2"} into unordered_map
+// ============================================================
+// 将扁平 JSON 对象中的字符串键值对解析到 unordered_map / Parse a flat JSON object of string key-value pairs into an unordered_map.
+static std::unordered_map<std::string, std::string> parse_headers_json(const char* json) {
+    std::unordered_map<std::string, std::string> headers;
+    if (!json || !*json) return headers;
+
+    std::string s(json);
+    // 极简状态机解析器，处理扁平的字符串键值对象 / Very simple state-machine parser for flat string-key/value objects
+    enum { OUTSIDE, IN_KEY, AFTER_KEY, IN_VALUE } state = OUTSIDE;
+    std::string current_key;
+    std::string current_value;
+
+    for (size_t i = 0; i < s.size(); ++i) {
+        char c = s[i];
+        switch (state) {
+        case OUTSIDE:
+            if (c == '"') { state = IN_KEY; current_key.clear(); }
+            break;
+        case IN_KEY:
+            if (c == '"') { state = AFTER_KEY; }
+            else if (c == '\\' && i + 1 < s.size()) { current_key += s[++i]; }
+            else { current_key += c; }
+            break;
+        case AFTER_KEY:
+            if (c == ':') { state = IN_VALUE; current_value.clear(); }
+            break;
+        case IN_VALUE:
+            if (c == '"') {
+                // 读取到闭合引号 / Read until closing quote
+                ++i;
+                while (i < s.size() && s[i] != '"') {
+                    if (s[i] == '\\' && i + 1 < s.size()) { current_value += s[++i]; }
+                    else { current_value += s[i]; }
+                    ++i;
+                }
+                headers[current_key] = current_value;
+                state = OUTSIDE;
+            }
+            break;
+        }
+    }
+    return headers;
+}
+
+// ============================================================
+// HTTP 客户端实现（改编自 dstalk_core HttpClient） / HTTP Client implementation (adapted from dstalk_core HttpClient)
+// ============================================================
+struct HttpClientCtx {
+    asio::io_context ioc;
+    ssl::context     ssl_ctx{ssl::context::tlsv12_client};
+    int              connect_timeout = 30;
+    int              request_timeout = 120;
+
+    HttpClientCtx() {
+        ssl_ctx.set_default_verify_paths();
+        // 启用对等证书验证 (CVSS 7.4 修复) / Enable peer certificate verification (CVSS 7.4 fix).
+        // set_default_verify_paths() 加载系统 CA 包；没有 verify_peer
+        // CA 存储不会被查询——任何证书（自签名/过期）都将被接受 / set_default_verify_paths() loads system CA bundle; without verify_peer
+        // the CA store is never consulted — any cert (self-signed/expired) is accepted.
+        // TODO: Windows: set_default_verify_paths() 可能无法定位系统 CA；
+        // 如果验证失败，设置 SSL_CERT_FILE 环境变量或捆绑 cacert.pem / Windows: set_default_verify_paths() may not locate system CAs;
+        // if verification fails, set SSL_CERT_FILE env or bundle a cacert.pem.
+        ssl_ctx.set_verify_mode(ssl::verify_peer);
+    }
+};
+
+// 核心 HTTP/HTTPS POST，支持可选 SSE 流式传输。执行 DNS 解析、
+// TLS 握手（含 SNI 和主机名验证），然后发送请求。
+// 如果 cb 非空，响应体将逐行解析用于流式传输 / Core HTTP/HTTPS POST with optional SSE streaming. Performs DNS resolve,
+// TLS handshake with SNI and hostname verification, then sends the request.
+// If `cb` is non-null, response body is parsed line-by-line for streaming.
+static int do_post_stream(
+    const char* host,
+    const char* port,
+    const char* target,
+    const char* body,
+    const char* headers_json,
+    dstalk_stream_cb cb,
+    void* userdata,
+    char** response_body,
+    int* status_code)
+{
+    if (!host || !port || !target || !body || !response_body || !status_code) {
+        if (response_body) *response_body = nullptr;
+        if (status_code) *status_code = -1;
+        return -1;
+    }
+
+    // 初始化输出 / Initialize output
+    *response_body = nullptr;
+    *status_code = -1;
+
+    // 从 C 回调构建 C++ lambda / Build C++ lambda from C callback
+    std::function<bool(const std::string&)> on_line;
+    if (cb) {
+        on_line = [cb, userdata](const std::string& line) -> bool {
+            return cb(line.c_str(), userdata) == 0;
+        };
+    }
+
+    HttpClientCtx ctx;
+
+    // 从配置读取超时设置 / Read timeouts from config if available
+    if (g_config_svc) {
+        const char* ct = g_config_svc->get("http.connect_timeout");
+        const char* rt = g_config_svc->get("http.request_timeout");
+        if (ct) ctx.connect_timeout = std::atoi(ct);
+        if (rt) ctx.request_timeout = std::atoi(rt);
+        if (ctx.connect_timeout <= 0) ctx.connect_timeout = 30;
+        if (ctx.request_timeout <= 0) ctx.request_timeout = 120;
+    }
+
+    std::string result_body;
+    int result_code = -1;
+
+    try {
+        tcp::resolver resolver(ctx.ioc);
+
+        // DNS 解析，10 秒超时。Boost.Asio 的同步 resolve()
+        // 在内部运行 io_context，因此定时器的 async_wait 回调在 resolve() 期间执行，
+        // 并在超时触发时调用 resolver.cancel() / DNS resolve with 10-second timeout.  Boost.Asio's synchronous
+        // resolve() runs the io_context internally, so the timer's async_wait
+        // callback executes during resolve() and calls resolver.cancel() when
+        // the deadline fires.
+        asio::steady_timer resolve_timer(ctx.ioc);
+        resolve_timer.expires_after(std::chrono::seconds(10));
+        resolve_timer.async_wait([&](const beast::error_code& ec) {
+            if (!ec) resolver.cancel();
+        });
+
+        beast::error_code resolve_ec;
+        auto endpoints = resolver.resolve(host, port, resolve_ec);
+        resolve_timer.cancel();
+
+        if (resolve_ec) {
+            if (g_host) g_host->log(DSTALK_LOG_ERROR,
+                "do_post_stream: DNS resolve %s:%s failed: %s",
+                host, port, resolve_ec.message().c_str());
+            result_body = std::string("DNS resolve failed: ") + resolve_ec.message();
+            goto done;
+        }
+
+        beast::ssl_stream<beast::tcp_stream> stream(ctx.ioc, ctx.ssl_ctx);
+        beast::flat_buffer buffer;
+
+        // SNI 主机名 / SNI hostname
+        if (!SSL_set_tlsext_host_name(stream.native_handle(), host)) {
+            if (g_host) g_host->log(DSTALK_LOG_ERROR,
+                "do_post_stream: SNI hostname set failed for %s", host);
+            result_body = "SNI hostname set failed";
+            goto done;
+        }
+
+        // 主机名验证：要求服务器证书 CN/SAN 匹配 'host'。
+        // 与 ssl::verify_peer 协同工作——没有它的话，
+        // 使用不同主机名的有效 CA 签名证书进行 MITM 攻击仍可通过 / Hostname verification: require server certificate CN/SAN to match
+        // 'host'.  This works in conjunction with ssl::verify_peer on the
+        // context — without it MITM with a valid CA-signed cert for a
+        // different hostname would still pass.
+        if (!SSL_set1_host(stream.native_handle(), host)) {
+            if (g_host) g_host->log(DSTALK_LOG_ERROR,
+                "do_post_stream: SSL_set1_host failed for %s", host);
+            result_body = "SSL_set1_host failed";
+            goto done;
+        }
+
+        // 连接 / Connect
+        beast::get_lowest_layer(stream).expires_after(
+            std::chrono::seconds(ctx.connect_timeout));
+        beast::get_lowest_layer(stream).connect(endpoints);
+        beast::get_lowest_layer(stream).expires_never();
+
+        // SSL 握手 / SSL handshake
+        beast::get_lowest_layer(stream).expires_after(
+            std::chrono::seconds(ctx.connect_timeout));
+        stream.handshake(ssl::stream_base::client);
+        beast::get_lowest_layer(stream).expires_never();
+
+        // 构建 HTTP POST 请求 / Build HTTP POST request
+        http::request<http::string_body> req{http::verb::post, target, 11};
+        req.set(http::field::host, host);
+        req.set(http::field::user_agent, "dstalk/0.1");
+        req.set(http::field::content_type, "application/json");
+        req.body() = body;
+        req.prepare_payload();
+
+        // 从 JSON 添加额外的头 / Add extra headers from JSON
+        auto extra_headers = parse_headers_json(headers_json);
+        for (const auto& h : extra_headers) {
+            req.set(h.first, h.second);
+        }
+
+        // 发送 / Send
+        beast::get_lowest_layer(stream).expires_after(
+            std::chrono::seconds(ctx.request_timeout));
+        http::write(stream, req);
+        beast::get_lowest_layer(stream).expires_never();
+
+        // 读取响应 / Read response
+        http::response_parser<http::string_body> parser;
+        parser.body_limit(16 * 1024 * 1024);
+        beast::get_lowest_layer(stream).expires_after(
+            std::chrono::seconds(ctx.request_timeout));
+        http::read_header(stream, buffer, parser);
+        beast::get_lowest_layer(stream).expires_never();
+
+        result_code = parser.get().result_int();
+
+        beast::error_code ec;
+
+        if (on_line) {
+            std::string fragment = parser.get().body();
+            auto emit_lines = [&]() -> bool {
+                size_t pos = 0;
+                while (pos < fragment.size()) {
+                    size_t nl = fragment.find('\n', pos);
+                    if (nl == std::string::npos) break;
+                    std::string line = fragment.substr(pos, nl - pos);
+                    if (!line.empty() && line.back() == '\r')
+                        line.pop_back();
+                    if (!on_line(line)) return false;
+                    pos = nl + 1;
+                }
+                if (pos > 0)
+                    fragment = fragment.substr(pos);
+                return true;
+            };
+            if (!emit_lines()) goto done;
+
+            size_t processed = parser.get().body().size();
+            while (!parser.is_done()) {
+                beast::get_lowest_layer(stream).expires_after(
+                    std::chrono::seconds(ctx.request_timeout));
+                http::read_some(stream, buffer, parser, ec);
+                if (ec) break;
+
+                const std::string& full_body = parser.get().body();
+                if (full_body.size() > processed) {
+                    std::string_view new_data(full_body.data() + processed,
+                                              full_body.size() - processed);
+                    processed = full_body.size();
+
+                    fragment.append(new_data.data(), new_data.size());
+                    if (!emit_lines()) goto done;
+                }
+            }
+            if (!fragment.empty()) {
+                if (fragment.back() == '\r')
+                    fragment.pop_back();
+                if (!fragment.empty())
+                    on_line(fragment);
+            }
+        } else {
+            while (!parser.is_done()) {
+                beast::get_lowest_layer(stream).expires_after(
+                    std::chrono::seconds(ctx.request_timeout));
+                http::read_some(stream, buffer, parser, ec);
+                if (ec) break;
+            }
+        }
+
+        result_body = parser.get().body();
+        beast::get_lowest_layer(stream).cancel();
+        stream.shutdown(ec);
+    } catch (const std::exception& e) {
+        if (g_host) g_host->log(DSTALK_LOG_ERROR,
+            "do_post_stream: %s", e.what());
+        result_code = -1;
+        result_body = e.what();
+    } catch (...) {
+        if (g_host) g_host->log(DSTALK_LOG_ERROR,
+            "do_post_stream: unknown exception (non-std::exception)");
+        result_code = -1;
+        result_body = "unknown exception";
+    }
+
+done:
+    *status_code = result_code;
+    if (!result_body.empty()) {
+        *response_body = g_host->strdup(result_body.c_str());
+    }
+    return (result_code >= 200 && result_code < 300) ? 0 : -1;
+}
+
+// ============================================================
+// 服务实现 / Service implementations
+// ============================================================
+// 同步 HTTP POST，返回完整响应体 / Synchronous HTTP POST returning the complete response body.
+static int http_post_json(
+    const char* host, const char* port,
+    const char* target, const char* body,
+    const char* headers_json,
+    char** response_body, int* status_code)
+{
+    return do_post_stream(host, port, target, body, headers_json,
+                          nullptr, nullptr, response_body, status_code);
+}
+
+// HTTP POST 带 SSE 流式传输：响应行到达时通过 cb 回调传递 / HTTP POST with SSE streaming: response lines are delivered to `cb` as they arrive.
+static int http_post_stream(
+    const char* host, const char* port,
+    const char* target, const char* body,
+    const char* headers_json,
+    dstalk_stream_cb cb, void* userdata,
+    char** response_body, int* status_code)
+{
+    return do_post_stream(host, port, target, body, headers_json,
+                          cb, userdata, response_body, status_code);
+}
+
+static dstalk_http_service_t g_service = {
+    http_post_json,
+    http_post_stream
+};
+
+// ============================================================
+// 插件生命周期 / Plugin lifecycle
+// ============================================================
+// 插件初始化：保存主机指针，查询 config 服务，注册 http 服务 / Plugin init: store host pointer, query config service, register http service.
+static int on_init(const dstalk_host_api_t* host) {
+    g_host = host;
+
+    // 查询 config 服务（声明的依赖） / Query config service (declared dependency)
+    g_config_svc = (dstalk_config_service_t*)host->query_service("config", 1);
+
+    return host->register_service("http", 1, &g_service);
+}
+
+// 插件关闭：无需清理 / Plugin shutdown: nothing to clean up.
+static void on_shutdown() {
+    // 无需清理 / nothing to clean up
+}
+
+static dstalk_plugin_info_t g_info = {
+    "http",                // name 名称
+    "1.0.0",               // version 版本
+    "HTTP/HTTPS client service using Boost.Beast + OpenSSL",  // description 描述
+    DSTALK_API_VERSION,    // api_version
+    {"config", nullptr},   // dependencies 依赖
+    on_init,               // on_init
+    on_shutdown,           // on_shutdown
+    nullptr                // on_event
+};
+
+// 必须入口点：返回插件描述符给主机 / Mandatory entry point: returns the plugin descriptor to the host.
+extern "C" DSTALK_PLUGIN_EXPORT dstalk_plugin_info_t* dstalk_plugin_init(void) {
+    return &g_info;
+}