diff --git a/gateway/README.md b/gateway/README.md index 09fdc86..e3236db 100644 --- a/gateway/README.md +++ b/gateway/README.md @@ -7,7 +7,8 @@ UniServe Framework 的应用层网关，C++20 实现。 ## 职责 - **账号鉴权**：`/register` 注册（需邮箱验证码，经 pymsg 调 smtp:8008 校验）、`/login` 登录，签发 RSA 签名的 Token。 -- **反向代理**：携带 Token 的请求校验后转发到 `Proxy.opened` 中放开的内部服务。 +- **HTTP 反向代理**：携带 Token 的请求校验后转发到 `Proxy.opened` 中放开的内部服务。 +- **WebSocket 反向代理（仅管理员）**：`[WsProxy]` 段配置的路径在 HTTP Upgrade 握手时校验 Token（`is_admin=true`），通过后透明桥接到后端 ws:// 或 wss:// 服务，支持文本/二进制帧双向转发。 - **CAPTCHA 接口**：`/captcha` 生成验证码 token + 文本（内存存储，5 分钟过期，一次性使用）。 - **邮箱验证码**：`/register/send_code` 发送 6 位验证码邮件（10 分钟有效）；注册时校验，一次性消费。 - **速率限制**：登录 5 次/5 分钟、注册 3 次/10 分钟、发验证码 5 次/10 分钟（IP 级）+ 3 次/小时/10 次/天（邮箱级），按 `X-Forwarded-For` 客户端 IP 计数（由 WAF 注入），超限一律返回 429。 @@ -124,6 +125,27 @@ Target-Url: http://127.0.0.1:9200 // 可选，缺省取 Proxy.default 返回 CORS 预检响应：`Access-Control-Allow-Origin: *`、`Allow-Methods: GET, POST, OPTIONS`、`Max-Age: 86400`。 +### WebSocket 代理（`[WsProxy]` 配置的路径） + +握手阶段鉴权规则： + +1. 从 `Authorization: Bearer ` 头读取 Token；若缺失则回退到 `Cookie: q4_token=`。 +2. RSA 验签 + 过期校验（与 HTTP 代理同样的 Token 机制）。 +3. Token payload 中必须 `is_admin=true`（即登录账号在 `Admin.users` 白名单内）。 +4. 鉴权失败 → 返回 HTTP 403 Forbidden，连接被拒绝。 +5. 鉴权通过 → 升级为 WebSocket 并透明桥接到配置的后端 ws:// 或 wss:// 地址，双向文本/二进制帧原样转发。 + +客户端使用方式与普通 WebSocket 一致，只需携带有效管理员 Token： + +```javascript +// 浏览器（同源会自动带 q4_token Cookie） +const ws = new WebSocket("wss://qsont.me/ws/aibox"); + +// 或显式带 Authorization（构造子协议或连接后首条消息均可，这里由网关在 Upgrade 头校验） +``` + +注意：WS 代理路由必须精确匹配（与 `ws()` 一致，非通配），且优先级高于 HTTP 通配 `*` 路由。 + ## 数据库 用户数据存于 db 服务（pymsg/TCP :8005）的 SQLite 库，表 `Users`（由 [db/create_sqlite3.sql](../db/create_sqlite3.sql) 建立，gateway 不再自动建表）： @@ -160,6 +182,11 @@ opened = http://127.0.0.1:9200;http://127.0.0.1:8006; # Token 有效期（秒），默认 3600 = 1 小时 keep_time = 3600 +[Admin] +# 管理员白名单（分号分隔邮箱），登录命中时 Token payload 写入 is_admin=true +# WebSocket 代理仅允许管理员访问，其他需管理员权限的下游同样凭此字段判断 +users = admin@example.com + [CORS] # 允许跨域的 Origin 白名单（分号分隔） # 仅命中的 Origin 才会回填 Access-Control-Allow-Origin；缺省为空（不回填） @@ -167,13 +194,22 @@ allowed_origins = https://qsont.me;https://www.qsont.me;http://127.0.0.1:8443;ht [Database] path = sqlite3.db + +[WsProxy] +# WebSocket 反向代理路由（仅管理员） +# 格式：/path=ws://host:port/target;/path2=wss://host2/target2 +# 支持 ws:// 和 wss:// 后端，握手阶段校验管理员 Token +# routes = /ws/aibox=ws://127.0.0.1:8765;/ws/log=wss://example.com/log +routes = ``` - `Proxy.default`：未带 `Target-Url` header 时的默认转发地址（不需 Token，主要服务前端静态资源）。 - `Proxy.opened`：允许外部带 Token 访问的内部服务白名单（分号分隔）。 - `Proxy.closed`：仅记录用途，实际靠"不在 opened 中即拒绝"实现隔离。 - `Token.keep_time`：登录签发 Token 的有效期（秒），缺省 3600。 +- `Admin.users`：管理员白名单（分号分隔邮箱），登录时命中即在 Token 中写入 `is_admin=true`。 - `CORS.allowed_origins`：OPTIONS 预检允许跨域的 Origin 白名单（分号分隔），缺省为空（不回填 ACAO）。 +- `WsProxy.routes`：WebSocket 代理路由映射，格式 `/path=ws://host:port/target`，多个用分号分隔；留空则不开启 WS 代理。 ## 部署模式 @@ -240,14 +276,15 @@ gateway/ ## 版本 -`0.5.4`（见 [src/main.cpp:31](src/main.cpp#L31)）。启动时打印： +`0.6.0`（见 [src/main.cpp:31](src/main.cpp#L31)）。启动时打印： ``` -Gateway 0.5.4 started +Gateway 0.6.0 started ``` 近期变更： +- `0.6.0`：WebSocket 反向代理（仅管理员）——新增 `[WsProxy]` 配置段，支持 `routes = /path=ws://host:port/target;...` 精确路径映射；握手阶段校验 Token 中 `is_admin=true`（基于 `Admin.users` 白名单），通过后双向透明桥接文本/二进制帧到后端，支持 `ws://` 和 `wss://`；后端连接错误/鉴权失败均优雅关闭。依赖 libhttp v3.1+ 的 `ws_proxy()` 接口。顺带修复 `LocalTime.hpp` 在 Apple clang 下 C++20 tzdb 不可用的编译问题（改用 `strftime`/`get_time`/`mktime`）。 - `0.5.4`：修复邮件轰炸漏洞（审计报告 P0-1）——`/register/send_code` 强制 CAPTCHA 校验（经 captcha:8009，Token 一次性消费），新增邮箱级双窗口限流（3 次/小时 + 10 次/天）；smtp 层同步增加邮箱级长期限流兜底 + 全局熔断（`SMTP_HOURLY_LIMIT` 环境变量，默认 100 封/小时）；前端 `register.html`/`register.js` 同步接入图形验证码（复用 `/captcha/image` 代理路由，方案 A：仅在「获取验证码」时要求 CAPTCHA，注册提交不加）。 - `0.5.3`：配合 smtp 同邮箱 60 秒发送冷却——`/register/send_code` 在 smtp 返回 `retry_after` 时按 429 透传其提示文案（其余发送失败仍用通用 500 文案，SMTP 内部错误细节不向外暴露）。 - `0.5.2`：限流错误码统一为 429（登录原 400、注册/发码原 403，并修正 "attampts" 拼写）；登录时用户不存在不再兜底 500——空结果统一走 401，并执行 dummy PBKDF2 校验拉平时序防用户名枚举；db 查询失败消息由误名的 "Login (Update Step) Failed" 改为 "Login (Query Step) Failed"；pymsg 下游调用超时常量化（`DbC::DEFAULT_TIMEOUT`、`_TIMEOUT_CAPTCHA/_SMTP_VERIFY/_SMTP_SEND`）。 diff --git a/gateway/example.ini b/gateway/example.ini index 9221437..371b5d6 100644 --- a/gateway/example.ini +++ b/gateway/example.ini @@ -32,3 +32,12 @@ allowed_origins = https://qsont.me;https://www.qsont.me;http://127.0.0.1:8443;ht [Database] path = sqlite3.db + +[WsProxy] +# WebSocket 反向代理路由（仅管理员可访问） +# 格式：routes = /path=ws://host:port/target;/path2=ws://host2:port2/target2 +# 握手阶段会校验 HTTP Upgrade 请求中的 Token（Authorization: Bearer xxx 或 Cookie: q4_token=xxx）， +# 仅 Token 中 is_admin=true 的用户才能升级连接；非管理员或无 Token 直接拒绝。 +# 支持 ws:// 和 wss:// 后端 +# 示例：routes = /ws/aibox=ws://127.0.0.1:8765;/ws/other=wss://example.com/ws +routes = diff --git a/gateway/inc/LocalTime.hpp b/gateway/inc/LocalTime.hpp index be7d415..2883845 100644 --- a/gateway/inc/LocalTime.hpp +++ b/gateway/inc/LocalTime.hpp @@ -1,20 +1,23 @@ #pragma once #include #include -#include #include #include +#include +#include namespace qing { class LocalTime { public: // 获取当前本地时间字符串 static std::string getCurrent(const std::string& format = "%Y-%m-%d %H:%M:%S") { - using namespace std::chrono; - auto now = system_clock::now(); - auto zone = current_zone(); - zoned_time zt{zone, now}; - return std::vformat("{:" + format + "}", std::make_format_args(zt)); + auto now = std::chrono::system_clock::now(); + std::time_t t = std::chrono::system_clock::to_time_t(now); + std::tm tm_buf{}; + localtime_r(&t, &tm_buf); + char buf[128]; + std::strftime(buf, sizeof(buf), format.c_str(), &tm_buf); + return std::string(buf); } // 解析时间字符串为 system_clock::time_point（本地时间 → UTC） @@ -22,29 +25,29 @@ public: const std::string& str, const std::string& format = "%Y-%m-%d %H:%M:%S" ) { - using namespace std::chrono; + std::tm tm_buf{}; std::istringstream iss{str}; - local_time lt; - iss >> std::chrono::parse(format, lt); // ✅ 修复 + iss >> std::get_time(&tm_buf, format.c_str()); if (iss.fail()) { return std::nullopt; } - auto zone = current_zone(); - zoned_time zt{zone, lt}; - return zt.get_sys_time(); + // mktime 将本地时间转换为 time_t(UTC) + std::time_t t = std::mktime(&tm_buf); + if (t == -1) { + return std::nullopt; + } + return std::chrono::system_clock::from_time_t(t); } // ----- 过期判断 ----- static bool isExpired(const std::chrono::system_clock::time_point& tp) { - using namespace std::chrono; - return system_clock::now() > tp; + return std::chrono::system_clock::now() > tp; } template static bool isExpired(const std::chrono::system_clock::time_point& tp, const std::chrono::duration& gracePeriod) { - using namespace std::chrono; - return system_clock::now() > tp + gracePeriod; + return std::chrono::system_clock::now() > tp + gracePeriod; } static bool isExpired(const std::string& timeStr, diff --git a/gateway/src/main.cpp b/gateway/src/main.cpp index cbd0262..2437b17 100644 --- a/gateway/src/main.cpp +++ b/gateway/src/main.cpp @@ -28,7 +28,7 @@ using namespace qing; namespace json = boost::json; namespace po = boost::program_options; -constexpr const char *_VER = "0.5.4"; +constexpr const char *_VER = "0.6.0"; ////////////////////////////////// #include @@ -339,6 +339,10 @@ static int _TOKEN_MAX_REFRESH = 10; // Token 生命周期内最大刷新次数 static std::vector _CORS_ORIGINS; // CORS 允许的 Origin 白名单，可由 [CORS] allowed_origins 配置 static std::vector _ADMIN_USERS; // 管理员账号白名单，可由 [Admin] users 配置（分号分隔），登录时写入 Token payload 的 is_admin +/* WebSocket 代理路由：path -> ws://backend-host:port/target + * 仅管理员用户可访问，握手阶段由 ws_proxy 回调验签 Token 中的 is_admin */ +static std::unordered_map _WS_PROXY_ROUTES; + /* instance items */ static std::unique_ptr _SERVE; static std::unique_ptr _KEYGEN; @@ -520,6 +524,31 @@ int main(int argc, char **argv) { _TOKEN_MAX_REFRESH = cf.get("Token.max_refresh", 10); _CORS_ORIGINS = split(cf.get("CORS.allowed_origins", ""), ';'); _ADMIN_USERS = split(cf.get("Admin.users", ""), ';'); + + /* [WsProxy] WebSocket 代理路由：格式 /path=ws://host:port/target;... + * 仅管理员（Token 中 is_admin=true）可访问，用于 aibox 等内部 WS 服务的网关暴露 */ + { + std::string ws_cfg = cf.get("WsProxy.routes", ""); + if (!ws_cfg.empty()) { + auto entries = split(ws_cfg, ';'); + for (auto& entry : entries) { + auto eq = entry.find('='); + if (eq != std::string::npos) { + std::string path = entry.substr(0, eq); + std::string url = entry.substr(eq + 1); + // 去除首尾空白 + auto trim = [](std::string& s) { + while (!s.empty() && std::isspace((unsigned char)s.front())) s.erase(s.begin()); + while (!s.empty() && std::isspace((unsigned char)s.back())) s.pop_back(); + }; + trim(path); trim(url); + if (!path.empty() && !url.empty()) { + _WS_PROXY_ROUTES[path] = url; + } + } + } + } + } } @@ -611,6 +640,81 @@ int main(int argc, char **argv) { //} +/* WebSocket 代理握手鉴权：从 HTTP Upgrade 请求中提取 Token 并验证 is_admin。 + * 返回 true 表示鉴权通过，out_target 填入后端 ws:// URL；返回 false 拒绝连接。 */ +static bool verifyWsAdminToken(const http::request& req, + const std::string& target_url, + std::string& out_target) { + out_target.clear(); + try { + std::string raw; + auto it = req.base().find(http::field::authorization); + if (it != req.base().end()) { + raw = std::string(it->value()); + } else { + auto cit = req.base().find(http::field::cookie); + if (cit != req.base().end()) { + std::string cookie_token = get_cookie_value(std::string(cit->value()), "q4_token"); + if (!cookie_token.empty()) + raw = "Bearer " + cookie_token; + } + } + if (raw.empty() || raw.substr(0, 7) != "Bearer ") + return false; + + raw = raw.substr(7); + size_t idx = raw.find("_"); + if (idx == std::string::npos) return false; + + std::string data = raw.substr(0, idx); + std::string sign = raw.substr(idx + 1); + std::string auth = Base64::base64_decode_to_string(data); + + auto publ = MyRSA::Public_Key::from_pem(_KEYGEN->get_public_key_pem()); + if (!publ.Verify(auth, sign)) return false; + + json::value jv_t; + try { jv_t = json::parse(auth); } + catch (...) { return false; } + if (!jv_t.is_object()) return false; + json::object& obj = jv_t.as_object(); + + auto t = json::value_to(obj["created_at"]); + auto keep_time_val = json::value_to(obj["keep_time"]); + auto created_tp = LocalTime::parse(t); + if (created_tp) { + auto now = std::chrono::system_clock::now(); + auto expire_tp = *created_tp + std::chrono::seconds(keep_time_val); + if (now > expire_tp) return false; + } + + // 必须是管理员 + bool is_admin = false; + if (obj.contains("is_admin")) { + auto& v = obj["is_admin"]; + if (v.is_bool()) is_admin = v.as_bool(); + } + if (!is_admin) { + logs().warning("gateway", "WS proxy rejected: non-admin user", + std::nullopt, std::string(req.target())); + return false; + } + + out_target = target_url; + // 将客户端请求路径附加到目标URL后（支持子路径） + std::string req_path = std::string(req.target()); + auto qpos = req_path.find('?'); + if (qpos != std::string::npos) req_path = req_path.substr(0, qpos); + // 不追加路径：代理直接使用配置中指定的完整目标URL + (void)req_path; + return true; + } + catch (const std::exception& e) { + std::cerr << "[WS proxy auth error] " << e.what() << std::endl; + return false; + } +} + /* 注册http路由 */ void registerRoutes(HttpServer &server) { @@ -666,6 +770,20 @@ void registerRoutes(HttpServer &server) { [](const auto& req, const auto& params) { return handleProxy(req, params, "post"); }); }); + /* WebSocket 反向代理（仅管理员）： + * ws_proxy 回调在 HTTP Upgrade 握手阶段触发， + * 返回非空 ws:// URL 表示鉴权通过并桥接到该后端；返回空串则拒绝连接。 */ + for (const auto& [path, url] : _WS_PROXY_ROUTES) { + server.ws_proxy(path, [url](const http::request& upgradeReq) -> std::string { + std::string target; + if (verifyWsAdminToken(upgradeReq, url, target)) { + return target; + } + return ""; + }); + std::cout << " WS-PROXY " << path << " -> " << url << " (admin only)" << std::endl; + } + } /* Http视图包装器 diff --git a/libhttp/README.md b/libhttp/README.md index 6316be6..0f01707 100644 --- a/libhttp/README.md +++ b/libhttp/README.md @@ -7,6 +7,7 @@ C++20 network library for the UniServe microservice framework, providing unified ### HttpServer - **HTTP/HTTPS Server**: Asynchronous HTTP 1.1 server with TLS support - **WebSocket Integration**: Single port can serve both HTTP(S) and WebSocket(S) endpoints +- **WebSocket Reverse Proxy**: Transparent bidirectional WebSocket proxy with handshake-time auth/callback (supports ws:// and wss:// backends) - **Routing**: Exact path matching + wildcard (`*`) routing - **Middleware**: `use()` for middleware chains (CORS, auth, logging, etc.) - **Static Files**: `setStaticDirectory()` for serving static files @@ -151,6 +152,28 @@ void disconnectWsClient(const std::string& clientId); size_t getWsClientCount() const; ``` +**WebSocket Reverse Proxy** +```cpp +void ws_proxy(const std::string& path, WsProxyHandshakeHandler handshake); +``` + +The `WsProxyHandshakeHandler` is called during the HTTP Upgrade handshake. It receives the upgrade request and must return the backend WebSocket URL (`ws://host:port/path` or `wss://host:port/path`) to bridge to. Returning an empty string rejects the connection (HTTP 403 is sent to the client). Proxy routes take priority over regular `ws()` routes on the same path. + +Example — admin-only proxy: +```cpp +server.ws_proxy("/ws/admin", [](const http::request& req) -> std::string { + if (!isAdmin(req)) return ""; // reject non-admins + return "ws://127.0.0.1:9000/stream"; // bridge to backend +}); +``` + +Features: +- Full-duplex async bridging (text and binary frames preserved) +- Write queues in both directions to maintain message ordering +- Automatic SNI for wss:// backends +- Graceful close propagation between client and backend +- Errors during backend connect/handshake close the client connection cleanly + **Lifecycle** ```cpp void start(); // non-blocking, spawns worker threads @@ -214,6 +237,9 @@ using WebSocketConnectHandler = std::function; + +using WsProxyHandshakeHandler = std::function& upgradeReq)>; ``` ## Backward Compatibility @@ -227,5 +253,6 @@ using WebSocketClientBase = HttpClient; ## Version History +- **v3.1.0**: Added WebSocket reverse proxy (`ws_proxy()`) with bidirectional async bridging, ws:// and wss:// backend support, handshake-time authorization callback. - **v3.0.0**: Merged WebSocketServer into HttpServer and WebSocketClient into HttpClient. Single-port HTTP+WebSocket support. - **v2.x**: Separate HttpServer, WebSocketServer, and HttpClient modules. diff --git a/libhttp/inc/us/HttpServer.hpp b/libhttp/inc/us/HttpServer.hpp index 843bd20..f2c07e4 100644 --- a/libhttp/inc/us/HttpServer.hpp +++ b/libhttp/inc/us/HttpServer.hpp @@ -53,6 +53,11 @@ using WebSocketConnectHandler = std::function; +// WebSocket 代理：握手回调接收 HTTP upgrade 请求，返回目标 ws:// 或 wss:// URL +// 返回空字符串表示拒绝（返回 403） +using WsProxyHandshakeHandler = std::function& upgradeReq)>; + struct RouteEntry { std::string path; std::regex pattern; @@ -65,6 +70,11 @@ struct WsRouteEntry { WebSocketMessageHandler handler; }; +struct WsProxyRouteEntry { + std::string path; + WsProxyHandshakeHandler handshake; +}; + class HttpServer { public: explicit HttpServer(const std::string& address, unsigned short port, @@ -87,6 +97,7 @@ public: // WebSocket 接口 void ws(const std::string& path, WebSocketMessageHandler handler); + void ws_proxy(const std::string& path, WsProxyHandshakeHandler handshake); void onWsConnect(WebSocketConnectHandler handler); void onWsDisconnect(WebSocketDisconnectHandler handler); bool sendToWsClient(const std::string& clientId, const std::string& message); @@ -139,6 +150,7 @@ private: // WebSocket 路由和处理 std::vector ws_handlers_; + std::vector ws_proxy_handlers_; WebSocketConnectHandler ws_connect_handler_; WebSocketDisconnectHandler ws_disconnect_handler_; std::unordered_map> ws_clients_; @@ -155,6 +167,7 @@ private: HttpRequestHandler findRoute(const std::string& method, const std::string& path); bool routeExists(const std::string& method, const std::string& path); WebSocketMessageHandler findWsRoute(const std::string& path); + WsProxyHandshakeHandler findWsProxyRoute(const std::string& path); // WebSocket 客户端管理 std::string generateWsClientId(); diff --git a/libhttp/src/HttpServer.cpp b/libhttp/src/HttpServer.cpp index a918262..aaa9580 100644 --- a/libhttp/src/HttpServer.cpp +++ b/libhttp/src/HttpServer.cpp @@ -35,14 +35,18 @@ class HttpServer::Session : public std::enable_shared_from_this { using ws_stream = websocket::stream; using wss_stream = websocket::stream; + // Proxy outbound types + using proxy_ws_stream = websocket::stream; + using proxy_wss_stream = websocket::stream&>; + public: Session(tcp::socket&& socket, HttpServer* server) : stream_(std::move(socket)), server_(server), ssl_enabled_(false), - is_websocket_(false), client_id_("") {} + is_websocket_(false), is_ws_proxy_(false), client_id_("") {} Session(tcp::socket&& socket, asio::ssl::context& ssl_ctx, HttpServer* server) : stream_(std::move(socket)), server_(server), ssl_enabled_(true), - is_websocket_(false), client_id_("") { + is_websocket_(false), is_ws_proxy_(false), client_id_("") { ssl_stream_ = std::make_unique(stream_, ssl_ctx); } @@ -78,6 +82,7 @@ public: private: void close() { beast::error_code ec; + closeProxy(websocket::close_code::going_away); if (is_websocket_) { if (ssl_enabled_ && wss_stream_) { wss_stream_->close(websocket::close_code::normal, ec); @@ -139,6 +144,33 @@ private: auto qpos = path.find('?'); if (qpos != std::string::npos) path = path.substr(0, qpos); + // Check proxy routes first (higher priority) + auto proxy_handshake = server_->findWsProxyRoute(path); + if (proxy_handshake) { + std::string target = proxy_handshake(request_); + if (target.empty()) { + // Rejected by handshake callback → 403 + auto resp = std::make_shared>(); + resp->version(request_.version()); + resp->result(http::status::forbidden); + resp->set(http::field::content_type, "text/plain"); + resp->set(http::field::connection, "close"); + resp->body() = "403 Forbidden\n"; + resp->prepare_payload(); + doWrite(resp); + return; + } + ws_path_ = path; + is_ws_proxy_ = true; + ws_proxy_target_ = target; + if (ssl_enabled_) { + doWssUpgrade(); + } else { + doWsUpgrade(); + } + return; + } + auto ws_handler = server_->findWsRoute(path); if (ws_handler) { ws_handler_ = ws_handler; @@ -217,6 +249,12 @@ private: std::cout << "[WS] Client " << client_id_ << " connected from " << remote_ip_ << ":" << remote_port_ << " at " << ws_path_ << std::endl; + if (is_ws_proxy_) { + std::cout << "[WS-Proxy] Proxying " << ws_path_ << " -> " << ws_proxy_target_ << std::endl; + doProxyConnect(ws_proxy_target_); + return; + } + if (server_->ws_connect_handler_) { server_->ws_connect_handler_(client_id_, getRemoteAddress()); } @@ -224,6 +262,8 @@ private: doWsRead(); } + // ======== Regular WebSocket message handling ======== + void doWsRead() { auto self = shared_from_this(); ws_buffer_.consume(ws_buffer_.size()); @@ -299,6 +339,7 @@ private: if (client_id_.empty()) return; std::cout << "[WS] Client " << client_id_ << " disconnected" << std::endl; + closeProxy(websocket::close_code::normal); server_->removeWsClient(client_id_); if (server_->ws_disconnect_handler_) { @@ -306,8 +347,361 @@ private: } client_id_.clear(); is_websocket_ = false; + is_ws_proxy_ = false; } + // ======== WebSocket Proxy (bidirectional bridge) ======== + + static void parseProxyUrl(const std::string& url, bool& ssl, std::string& host, + std::string& port, std::string& path) { + std::string rest; + if (url.size() >= 6 && url.substr(0, 6) == "wss://") { + ssl = true; + rest = url.substr(6); + } else if (url.size() >= 5 && url.substr(0, 5) == "ws://") { + ssl = false; + rest = url.substr(5); + } else { + ssl = false; + rest = url; + } + + size_t path_pos = rest.find('/'); + std::string host_port; + if (path_pos != std::string::npos) { + host_port = rest.substr(0, path_pos); + path = rest.substr(path_pos); + } else { + host_port = rest; + path = "/"; + } + + size_t port_pos = host_port.find(':'); + if (port_pos != std::string::npos) { + host = host_port.substr(0, port_pos); + port = host_port.substr(port_pos + 1); + } else { + host = host_port; + port = ssl ? "443" : "80"; + } + } + + void doProxyConnect(const std::string& target_url) { + auto self = shared_from_this(); + bool ssl; + std::string host, port, path; + parseProxyUrl(target_url, ssl, host, port, path); + proxy_ssl_ = ssl; + proxy_host_ = host; + proxy_port_ = port; + proxy_path_ = path; + + auto exec = stream_.get_executor(); + proxy_resolver_ = std::make_unique(asio::make_strand(exec)); + + proxy_resolver_->async_resolve(host, port, + asio::bind_executor(exec, + [self](beast::error_code ec, tcp::resolver::results_type results) { + self->onProxyResolve(ec, results); + })); + } + + void onProxyResolve(beast::error_code ec, tcp::resolver::results_type results) { + if (ec) { + std::cerr << "[WS-Proxy] Resolve error: " << ec.message() << std::endl; + close(); + return; + } + + auto self = shared_from_this(); + auto exec = stream_.get_executor(); + + proxy_socket_ = std::make_unique(exec); + proxy_tcp_stream_ = std::make_unique(std::move(*proxy_socket_)); + proxy_socket_.reset(); + + proxy_tcp_stream_->expires_after(std::chrono::seconds(10)); + asio::async_connect(proxy_tcp_stream_->socket(), results, + asio::bind_executor(exec, + [self](beast::error_code ec, tcp::endpoint ep) { + self->onProxyConnect(ec, ep); + })); + } + + void onProxyConnect(beast::error_code ec, tcp::endpoint /*ep*/) { + if (ec) { + std::cerr << "[WS-Proxy] Connect error: " << ec.message() << std::endl; + close(); + return; + } + + auto self = shared_from_this(); + if (proxy_ssl_) { + proxy_ssl_ctx_ = std::make_unique(asio::ssl::context::tlsv12_client); + proxy_ssl_ctx_->set_verify_mode(asio::ssl::verify_none); + proxy_ssl_stream_ = std::make_unique>( + proxy_tcp_stream_->socket(), *proxy_ssl_ctx_); + + if (!SSL_set_tlsext_host_name(proxy_ssl_stream_->native_handle(), proxy_host_.c_str())) { + std::cerr << "[WS-Proxy] Failed to set SNI" << std::endl; + close(); + return; + } + + proxy_tcp_stream_->expires_after(std::chrono::seconds(10)); + auto exec = stream_.get_executor(); + proxy_ssl_stream_->async_handshake(asio::ssl::stream_base::client, + asio::bind_executor(exec, + [self](beast::error_code ec) { + self->onProxySslHandshake(ec); + })); + } else { + proxy_ws_stream_ = std::make_unique(proxy_tcp_stream_->socket()); + proxy_tcp_stream_->expires_never(); + + proxy_ws_stream_->set_option(websocket::stream_base::decorator( + [](websocket::request_type& req) { + req.set(http::field::user_agent, "UniServe-WsProxy"); + })); + + auto exec = self->stream_.get_executor(); + proxy_ws_stream_->async_handshake(proxy_host_, proxy_path_, + asio::bind_executor(exec, + [self](beast::error_code ec) { + self->onProxyHandshake(ec); + })); + } + } + + void onProxySslHandshake(beast::error_code ec) { + if (ec) { + std::cerr << "[WS-Proxy] SSL handshake error: " << ec.message() << std::endl; + close(); + return; + } + + auto self = shared_from_this(); + proxy_wss_stream_ = std::make_unique(*proxy_ssl_stream_); + proxy_tcp_stream_->expires_never(); + + proxy_wss_stream_->set_option(websocket::stream_base::decorator( + [](websocket::request_type& req) { + req.set(http::field::user_agent, "UniServe-WsProxy"); + })); + + auto exec = stream_.get_executor(); + proxy_wss_stream_->async_handshake(proxy_host_, proxy_path_, + asio::bind_executor(exec, + [self](beast::error_code ec) { + self->onProxyHandshake(ec); + })); + } + + void onProxyHandshake(beast::error_code ec) { + if (ec) { + std::cerr << "[WS-Proxy] Backend handshake error: " << ec.message() << std::endl; + close(); + return; + } + + proxy_connected_ = true; + proxy_closed_ = false; + std::cout << "[WS-Proxy] Connected to backend " << proxy_host_ << ":" << proxy_port_ + << proxy_path_ << std::endl; + + if (server_->ws_connect_handler_) { + server_->ws_connect_handler_(client_id_, getRemoteAddress()); + } + + // Start bidirectional bridging + doProxyReadClient(); + doProxyReadBackend(); + } + + // ---- Client → Backend direction ---- + + void doProxyReadClient() { + if (proxy_closed_) return; + auto self = shared_from_this(); + proxy_client_buffer_.consume(proxy_client_buffer_.size()); + + auto read_handler = [self](beast::error_code ec, std::size_t bytes_transferred) { + self->onProxyReadClient(ec, bytes_transferred); + }; + + if (ssl_enabled_ && wss_stream_) { + wss_stream_->async_read(proxy_client_buffer_, read_handler); + } else if (ws_stream_) { + ws_stream_->async_read(proxy_client_buffer_, read_handler); + } + } + + void onProxyReadClient(beast::error_code ec, std::size_t bytes_transferred) { + if (ec) { + if (ec != websocket::error::closed) { + std::cerr << "[WS-Proxy] Client read error: " << ec.message() << std::endl; + } + closeProxy(websocket::close_code::normal); + return; + } + + bool is_binary = false; + if (ssl_enabled_ && wss_stream_) { + is_binary = wss_stream_->got_binary(); + } else if (ws_stream_) { + is_binary = ws_stream_->got_binary(); + } + + std::string data = beast::buffers_to_string(proxy_client_buffer_.data()); + proxy_client_buffer_.consume(bytes_transferred); + + // Queue to backend + auto exec = stream_.get_executor(); + asio::post(exec, [self = shared_from_this(), data = std::move(data), is_binary]() { + self->proxy_to_backend_queue_.emplace_back(std::move(data), is_binary); + if (self->proxy_to_backend_queue_.size() == 1) { + self->doProxyWriteBackend(); + } + }); + + doProxyReadClient(); + } + + void doProxyWriteBackend() { + if (proxy_to_backend_queue_.empty() || !proxy_connected_ || proxy_closed_) return; + + auto self = shared_from_this(); + auto& [msg, is_binary] = proxy_to_backend_queue_.front(); + + auto write_handler = [self](beast::error_code ec, std::size_t) { + if (ec) { + std::cerr << "[WS-Proxy] Backend write error: " << ec.message() << std::endl; + self->closeProxy(websocket::close_code::internal_error); + return; + } + self->proxy_to_backend_queue_.pop_front(); + if (!self->proxy_to_backend_queue_.empty()) { + self->doProxyWriteBackend(); + } + }; + + if (proxy_ssl_ && proxy_wss_stream_) { + proxy_wss_stream_->binary(is_binary); + proxy_wss_stream_->async_write(asio::buffer(msg), write_handler); + } else if (proxy_ws_stream_) { + proxy_ws_stream_->binary(is_binary); + proxy_ws_stream_->async_write(asio::buffer(msg), write_handler); + } + } + + // ---- Backend → Client direction ---- + + void doProxyReadBackend() { + if (proxy_closed_) return; + auto self = shared_from_this(); + proxy_backend_buffer_.consume(proxy_backend_buffer_.size()); + + auto read_handler = [self](beast::error_code ec, std::size_t bytes_transferred) { + self->onProxyReadBackend(ec, bytes_transferred); + }; + + if (proxy_ssl_ && proxy_wss_stream_) { + proxy_wss_stream_->async_read(proxy_backend_buffer_, read_handler); + } else if (proxy_ws_stream_) { + proxy_ws_stream_->async_read(proxy_backend_buffer_, read_handler); + } + } + + void onProxyReadBackend(beast::error_code ec, std::size_t bytes_transferred) { + if (ec) { + if (ec != websocket::error::closed) { + std::cerr << "[WS-Proxy] Backend read error: " << ec.message() << std::endl; + } + closeProxy(websocket::close_code::normal); + return; + } + + bool is_binary = false; + if (proxy_ssl_ && proxy_wss_stream_) { + is_binary = proxy_wss_stream_->got_binary(); + } else if (proxy_ws_stream_) { + is_binary = proxy_ws_stream_->got_binary(); + } + + std::string data = beast::buffers_to_string(proxy_backend_buffer_.data()); + proxy_backend_buffer_.consume(bytes_transferred); + + // Queue to client + auto exec = stream_.get_executor(); + asio::post(exec, [self = shared_from_this(), data = std::move(data), is_binary]() { + self->proxy_to_client_queue_.emplace_back(std::move(data), is_binary); + if (self->proxy_to_client_queue_.size() == 1) { + self->doProxyWriteClient(); + } + }); + + doProxyReadBackend(); + } + + void doProxyWriteClient() { + if (proxy_to_client_queue_.empty() || !proxy_connected_ || proxy_closed_) return; + + auto self = shared_from_this(); + auto& [msg, is_binary] = proxy_to_client_queue_.front(); + + auto write_handler = [self](beast::error_code ec, std::size_t) { + if (ec) { + std::cerr << "[WS-Proxy] Client write error: " << ec.message() << std::endl; + self->closeProxy(websocket::close_code::internal_error); + return; + } + self->proxy_to_client_queue_.pop_front(); + if (!self->proxy_to_client_queue_.empty()) { + self->doProxyWriteClient(); + } + }; + + if (ssl_enabled_ && wss_stream_) { + wss_stream_->binary(is_binary); + wss_stream_->async_write(asio::buffer(msg), write_handler); + } else if (ws_stream_) { + ws_stream_->binary(is_binary); + ws_stream_->async_write(asio::buffer(msg), write_handler); + } + } + + void closeProxy(websocket::close_code code) { + if (proxy_closed_) return; + proxy_closed_ = true; + proxy_connected_ = false; + + beast::error_code ec; + // Close backend connection + if (proxy_ssl_ && proxy_wss_stream_) { + proxy_wss_stream_->close(code, ec); + } else if (proxy_ws_stream_) { + proxy_ws_stream_->close(code, ec); + } + if (proxy_ssl_stream_) { + proxy_ssl_stream_->shutdown(ec); + } + if (proxy_tcp_stream_) { + proxy_tcp_stream_->socket().shutdown(tcp::socket::shutdown_send, ec); + proxy_tcp_stream_->close(); + } + + proxy_to_backend_queue_.clear(); + proxy_to_client_queue_.clear(); + proxy_ws_stream_.reset(); + proxy_wss_stream_.reset(); + proxy_ssl_stream_.reset(); + proxy_ssl_ctx_.reset(); + proxy_tcp_stream_.reset(); + proxy_resolver_.reset(); + } + + // ======== Regular HTTP ======== + void processRequest(http::request req) { try { auto resp = std::make_shared>( @@ -352,6 +746,7 @@ private: } } + // ---- Client stream (inbound) ---- tcp_stream stream_; std::unique_ptr ssl_stream_; std::unique_ptr ws_stream_; @@ -368,10 +763,31 @@ private: unsigned short remote_port_ = 0; bool is_websocket_; + bool is_ws_proxy_; std::string client_id_; std::string ws_path_; WebSocketMessageHandler ws_handler_; std::deque ws_send_queue_; + + // ---- Proxy state (outbound to backend) ---- + std::string ws_proxy_target_; + bool proxy_ssl_ = false; + bool proxy_connected_ = false; + bool proxy_closed_ = false; + std::string proxy_host_, proxy_port_, proxy_path_; + + std::unique_ptr proxy_resolver_; + std::unique_ptr proxy_socket_; + std::unique_ptr proxy_tcp_stream_; + std::unique_ptr proxy_ssl_ctx_; + std::unique_ptr> proxy_ssl_stream_; + std::unique_ptr proxy_ws_stream_; + std::unique_ptr proxy_wss_stream_; + + beast::flat_buffer proxy_client_buffer_; + beast::flat_buffer proxy_backend_buffer_; + std::deque> proxy_to_backend_queue_; + std::deque> proxy_to_client_queue_; }; // ===================== Listener ===================== @@ -624,6 +1040,13 @@ void HttpServer::ws(const std::string& path, WebSocketMessageHandler handler) { ws_handlers_.push_back(entry); } +void HttpServer::ws_proxy(const std::string& path, WsProxyHandshakeHandler handshake) { + WsProxyRouteEntry entry; + entry.path = path; + entry.handshake = handshake; + ws_proxy_handlers_.push_back(entry); +} + void HttpServer::onWsConnect(WebSocketConnectHandler handler) { ws_connect_handler_ = handler; } @@ -738,6 +1161,13 @@ WebSocketMessageHandler HttpServer::findWsRoute(const std::string& path) { return nullptr; } +WsProxyHandshakeHandler HttpServer::findWsProxyRoute(const std::string& path) { + for (const auto& e : ws_proxy_handlers_) { + if (e.path == path) return e.handshake; + } + return nullptr; +} + // ---- WebSocket 客户端管理 ---- std::string HttpServer::generateWsClientId() { return "ws_client_" + std::to_string(++ws_client_counter_); @@ -924,6 +1354,9 @@ void HttpServer::showStatus() const { std::cout << " WebSocket: " << ws_handlers_.size() << " route(s)" << std::endl; for (const auto& entry : ws_handlers_) std::cout << " WS " << entry.path << std::endl; + std::cout << " WebSocket-Proxy: " << ws_proxy_handlers_.size() << " route(s)" << std::endl; + for (const auto& entry : ws_proxy_handlers_) + std::cout << " WS-PROXY " << entry.path << std::endl; std::cout << " WebSocket clients: " << getWsClientCount() << std::endl; std::cout << "Middlewares: " << middlewares_.size() << std::endl; std::cout << "==========================" << std::endl;