Compare commits
5
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 qing
|
e7a2877422 | ||
|
qing
|
7d6ba6de93 | ||
|
qing
|
b30ce1424b | ||
|
qing
|
7ef6a0b65f | ||
|
qing
|
bd1d7ca4d0 |
No files matched your search
@@ -10,3 +10,7 @@ timeout = 10
|
||||
[ai_review]
|
||||
url = http://127.0.0.1:8007/review
|
||||
timeout = 10
|
||||
|
||||
[gateway]
|
||||
url = http://127.0.0.1:8443/whoami
|
||||
timeout = 5
|
||||
+65
-21
@@ -14,13 +14,16 @@ Dendrachy Network File-System，基于父子关系（pid）组织的树状节点
|
||||
▼
|
||||
dnfs :8006 (Flask + Waitress)
|
||||
│
|
||||
├──► db :8005 数据持久化（SQL-over-HTTP，/execute）
|
||||
└──► ai_review :8007 内容审核（仅 add_note / update_note 时调用）
|
||||
├──► confs :8010 启动时拉取配置（dnfs.ini，强依赖，retry_on_refused 等待就绪）
|
||||
├──► db :8005 数据持久化（pymsg TCP RPC，execute 路由）
|
||||
└──► ai_review :8007 内容审核（HTTP POST /review，仅 add_note/update_note 时调用）
|
||||
```
|
||||
|
||||
- 不直连数据库，所有数据操作通过 HTTP 调用 `db` 服务（`http://127.0.0.1:8005/execute`）。
|
||||
- 新增/更新节点前调用 `ai_review` 服务（`http://127.0.0.1:8007/review`）进行内容审核，审核不通过则拒绝写入。
|
||||
- 通常部署在 gateway 之后（`Proxy.opened` 白名单中），由 gateway 完成 Token 验签后转发，dnfs 仅做所有权校验。
|
||||
- 启动时通过 pymsg 从 confs 拉取 `dnfs.ini` 配置，拉取失败直接退出（强依赖）；使用 `retry_on_refused=True` 自动等待 confs 就绪。
|
||||
- 不直连数据库，所有数据操作通过 pymsg TCP RPC 调用 `db` 服务（`_route: "execute"`），规避 SQL 注入风险。
|
||||
- 新增/更新节点前调用 `ai_review` 服务进行内容审核（HTTP POST `{content: "..."}`），审核不通过则拒绝写入。
|
||||
- 日志通过 pymsg `AsyncRpcQueue` 异步批量写入 db 的 `Logs` 表，fail-safe 不阻塞业务。
|
||||
- 通常部署在 gateway/WAF 之后（`Proxy.opened` 白名单中），由 gateway 完成 Token RSA 验签后转发。
|
||||
|
||||
## 数据模型
|
||||
|
||||
@@ -42,7 +45,7 @@ Dendrachy Network File-System，基于父子关系（pid）组织的树状节点
|
||||
Authorization: Bearer <base64_payload>_<rsa_sign>
|
||||
```
|
||||
|
||||
- Token 由 gateway 在 `/login` 签发，dnfs 通过 [dnfs.py:163](dnfs.py#L163) `decode_token` 仅解 Base64 取 payload，**不验签**（验签由 gateway 完成）。
|
||||
- Token 由 gateway 在 `/login` 签发，dnfs 通过 [dnfs.py:304](dnfs.py#L304) `decode_token` 仅解 Base64 取 payload，**当前代码不验签**（代码中标注 `# FIXME: 验证数字签名`，依赖部署上经 gateway 转发的网络隔离）。
|
||||
- payload 中取 `user` 字段作为操作者，写入 `owner`。
|
||||
- `delete_note` 与 `update_note` 校验所有权：
|
||||
- 节点 `owner` 必须等于当前 `user`，否则 `403`；
|
||||
@@ -208,10 +211,13 @@ curl -X POST http://localhost:8006/update_note \
|
||||
|
||||
## 安全特性
|
||||
|
||||
- **安全响应头**（[security_middleware.py](security_middleware.py)）：`X-Frame-Options: DENY`、`X-Content-Type-Options: nosniff`、`Referrer-Policy`、`Permissions-Policy`、CSP（`default-src 'self'`、`frame-ancestors 'none'`），并移除 `Server` 头。
|
||||
- **安全响应头**（[security_middleware.py](security_middleware.py)）：`X-Frame-Options: SAMEORIGIN`、`X-XSS-Protection`、`X-Content-Type-Options: nosniff`、`Referrer-Policy: strict-origin-when-cross-origin`、`Permissions-Policy`（禁用摄像头/麦克风/地理位置）、CSP（`default-src 'self'` 等），并移除 `Server` 头。
|
||||
- **Token 鉴权**：写操作强制校验，payload 取 `user` 字段。
|
||||
- **所有权隔离**：普通用户只能改/删自己的节点。
|
||||
- **内容审核**：写入与更新前调用 `ai_review`，色情/违禁/赌博/诈骗/暴力等内容会被拒绝。
|
||||
- **特殊节点管控**：link/html/image/file 类型有格式校验与大小限制（URL 前缀、文件名长度、内容大小上限）。
|
||||
- **内容审核**：写入与更新前调用 `ai_review`，非法内容会被拒绝（fail-closed 设计，见已知问题）。
|
||||
- **异步日志**：通过 pymsg `AsyncRpcQueue` 异步批量写日志，fail-safe 不阻塞业务线程。
|
||||
- **非 root 容器化运行**：Dockerfile 中使用 appuser 非特权用户运行服务。
|
||||
|
||||
## 配置
|
||||
|
||||
@@ -266,17 +272,17 @@ dnfs 通过 pymsg `retry_on_refused=True` 自动阻塞等待 confs 就绪，**
|
||||
|
||||
## 前端
|
||||
|
||||
三文件架构：
|
||||
前端页面由 `static` 服务承载，不在 dnfs 模块内：
|
||||
|
||||
```
|
||||
templates/index.html 页面结构（Font Awesome 图标）
|
||||
static/css/style.css 样式
|
||||
static/js/script.js 交互逻辑
|
||||
static/public/dnfs.html 页面结构
|
||||
static/public/css/ 样式（index.css / aibox.css 等）
|
||||
static/public/js/dnfs.js 交互逻辑
|
||||
```
|
||||
|
||||
功能：节点树浏览、新增/编辑/删除节点、文件 Base64 上传与下载、节点迁移（移动到其他父节点下）。
|
||||
|
||||
> 注：首页路由 `@app.route('/')` 当前已注释，前端页面实际由 `static` 服务的 `dnfs.html` 承载。
|
||||
> 注：首页路由 `@app.route('/')` 和 `render_template` 相关代码已注释，dnfs 仅提供 JSON API，不直接提供 HTML 页面。
|
||||
|
||||
## 依赖
|
||||
|
||||
@@ -284,12 +290,12 @@ static/js/script.js 交互逻辑
|
||||
flask
|
||||
flask-cors
|
||||
waitress
|
||||
nuitka # 可选，编译为独立二进制
|
||||
pymsg # 内部 RPC 通信库（../pymsg/）
|
||||
```
|
||||
|
||||
> `requirements.txt` 仍列有 `flask_sqlalchemy`，但代码已不使用（SQLAlchemy 直连代码已注释），属遗留依赖。
|
||||
运行时需要 `confs`（:8010，启动强依赖）、`db`（:8005）和 `ai_review`（:8007）服务可达；对外访问需经 gateway 转发并验签。
|
||||
|
||||
运行时需要 `db`（:8005）和 `ai_review`（:8007）服务已启动；对外访问需经 gateway 转发并验签。
|
||||
> `requirements.txt` 中曾列有 `flask_sqlalchemy`，Dockerfile 中已添加 `pip uninstall -y flask_sqlalchemy` 清理遗留依赖；代码中的 SQLAlchemy 直连逻辑已全部注释，通过 pymsg RPC 调用 db 服务。
|
||||
|
||||
## 运行
|
||||
|
||||
@@ -300,8 +306,46 @@ bash _run.sh # 启动服务（waitress）
|
||||
|
||||
## 已知问题与风险
|
||||
|
||||
- **`decode_token` 未验签**（[dnfs.py:167](dnfs.py#L167)）：仅解 Base64 取 payload，存在 `# FIXME: 验证数字签名` 标记。当前安全性依赖"必须经 gateway 转发"这一部署假设；若 dnfs 被直接访问（绕过 gateway），任意伪造的 Token 都能通过。建议补 RSA 验签或在部署上确保仅 gateway 可达。
|
||||
- **`review_content` 返回类型不一致**（[dnfs.py:73-119](dnfs.py#L73)）：正常路径返回 `bool`（[L105](dnfs.py#L105)），HTTPError 路径返回 `dict`（[L113](dnfs.py#L113)）。调用处用 `True != review` 判断，HTTPError 时 `dict != True` 恒为真，会一律判为审核失败并吞掉具体错误信息。建议统一返回类型或在调用处 catch 异常。
|
||||
- **CORS 全开**：`CORS(app)` 未配置白名单，允许任意 Origin 跨域访问。考虑到写操作需 Token，风险可控，但仍建议按 gateway 的 `[CORS] allowed_origins` 方式收紧。
|
||||
- **`call_db_service` 异常未捕获**：调用处（[L208](dnfs.py#L208)、[L261](dnfs.py#L261)、[L318](dnfs.py#L318)、[L347](dnfs.py#L347)、[L440](dnfs.py#L440)）直接取 `result['success']` / `notes['data']`，若 db 服务返回非预期结构或网络异常会抛 500。
|
||||
- **`get_notes` 不鉴权**：任何能访问 dnfs 的人都能枚举所有节点内容。若内容含敏感 Base64 文件，需在 gateway 层做路径级鉴权。
|
||||
审计日期: 2026-08-01
|
||||
|
||||
### 🔴 P0 — 严重问题（需立即修复）
|
||||
|
||||
1. **Token 无 RSA 签名验证，可伪造任意用户身份（含管理员）** — [dnfs.py:304-311](dnfs.py#L304-L311)
|
||||
- `decode_token()` 直接 base64 解码 payload，完全跳过 RSA 签名验证（代码中有 `# FIXME: 验证数字签名` 注释）
|
||||
- 若 dnfs 被直接访问（绕过 gateway），攻击者可构造 `{"user": "任意邮箱", "is_admin": true}` 的 payload，base64 编码后拼接下划线+任意字符串即可伪造管理员身份，绕过所有权限校验
|
||||
- **参考实现**: gateway `verifyWsAdminToken` 展示了正确流程：分割 `_` → base64 解码 → RSA 公钥验签 → 检查过期时间
|
||||
- **修复建议**: 实现 RSA 签名验证（持有 gateway 公钥）+ Token 过期校验（`iat`/`created_at` + `keep_time`）
|
||||
|
||||
2. **`review_content()` 错误处理断裂，fail-closed 失效** — [dnfs.py:149-195](dnfs.py#L149-L195)
|
||||
- 成功路径返回 `bool`，但 HTTPError/URLError/其他异常直接 `raise e`，没有返回 `False` 实现 fail-closed
|
||||
- 内层 `except json.JSONDecodeError as e` 变量名遮蔽外层 `HTTPError as e`
|
||||
- ai_review 不可用时异常向上传播导致 500，而非拒绝内容（违反 fail-closed 设计意图）
|
||||
- **修复建议**: 所有异常路径返回 `False` 或抛 DnfsExp 拒绝写入
|
||||
|
||||
3. **`delete_note` 和 `update_note` 缺少 `@_handle_exp` 装饰器** — [dnfs.py:428](dnfs.py#L428), [dnfs.py:531](dnfs.py#L531)
|
||||
- `add_note`/`get_notes` 使用了 `@_handle_exp` 统一异常处理，但 `delete_note`/`update_note` 没有
|
||||
- 抛出 `DnfsExp` 或其他异常时返回 Flask 默认 500 HTML 错误页，与其他接口 JSON 响应不一致
|
||||
|
||||
4. **有子节点的父节点可能被删除（树结构完整性保护缺陷）**
|
||||
- **数据库层缺失外键约束**: 虽然 dbv2 启用了 `PRAGMA foreign_keys=ON`，但 Notes 表建表 SQL（[db/create_sqlite3.sql:13-19](../db/create_sqlite3.sql#L13-L19)）中 `pid` 列只是普通 INTEGER，没有定义 `FOREIGN KEY (pid) REFERENCES Notes(id) ON DELETE RESTRICT`
|
||||
- **应用层 TOCTOU 竞态**: 子节点查询（SELECT）和删除（DELETE）是两个独立 RPC 调用，无事务保护，检查与删除之间可插入子节点导致父节点被删后子节点成为孤儿
|
||||
- **错误处理缺陷**: `_call_dbv2` 返回 `success=False` 时响应中无 `data` 键，直接访问 `notes['data']` 会抛 KeyError 而非正确处理
|
||||
- **修复建议**: (a) 添加数据库外键约束; (b) 应用层检查返回值 `success` 状态; (c) 使用数据库事务将检查+删除原子化
|
||||
|
||||
### 🟠 P1 — 高危问题
|
||||
|
||||
5. **Token 无过期校验** — payload 中包含 `iat`/`created_at`/`keep_time` 字段，但 `decode_token()` 完全不检查，泄露的 Token 可永久使用
|
||||
6. **请求体大小无限制** — Waitress 未配置 `max_request_body_size`，可能被大请求体 DoS；建议设置 10MB 上限与 WAF 对齐
|
||||
7. **`log_client.py` 配置与主模块不一致** — [log_client.py:32-33](log_client.py#L32-L33) 直接读 `DB_HOST`/`DB_PORT` 环境变量，不从 confs 拉取配置，可能连错地址
|
||||
8. **`update_note` 修改 pid 时未验证目标父节点存在** — `add_note` 会检查 pid≠0 时父节点是否存在，但 `update_note` 直接更新不检查，可将节点移动到不存在的父节点下
|
||||
9. **db RPC 客户端未启用启动等待** — `_rpc = RpcClient(...)` 未设置 `retry_on_refused=True`，db 未就绪时首次请求失败（虽然 confs 连接启用了重试）
|
||||
10. **`delete_note`/`update_note` 中 `decode_token` 无 try-except** — `get_notes` 中包裹了异常处理，但这两个函数没有，base64 解码/JSON 解析错误直接抛 500
|
||||
11. **`note_id` 缺少强制类型转换** — `note_id: int = data.get('id')` 只是类型提示，不会实际转换；传入字符串 `"5"` 依赖 SQLite 类型亲和性隐式匹配
|
||||
|
||||
### 🟡 P2 — 中危问题
|
||||
|
||||
12. **错误响应泄露内部信息** — 多处错误消息拼接 `request.remote_addr` 和用户名返回给客户端（如"未授权访问，从xxx"），应仅记录到服务端日志
|
||||
13. **CORS 配置过于宽松** — `CORS(app)` 默认允许所有来源（`*`），建议限制为实际前端域名
|
||||
14. **`_call_dbv2` 返回值访问前未统一检查 success** — 多处直接访问 `notes['data']` 而不先确认 `success==True`
|
||||
15. **`pid` 参数判断逻辑不严谨** — `if not pid: pid = 0` 会将空字符串/None 等非法输入默默转为 0，掩盖参数错误
|
||||
16. **遗留注释代码** — 文件中有大量注释掉的旧版 SQLAlchemy 代码和已废弃接口，影响可读性
|
||||
+49
-11
@@ -19,7 +19,6 @@ from waitress import serve
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import base64
|
||||
import configparser
|
||||
import urllib.request
|
||||
import urllib.error
|
||||
@@ -48,6 +47,7 @@ _dnfs_config = {
|
||||
'server': {'host': '', 'port': ''},
|
||||
'database': {'host': '', 'port': '', 'timeout': ''},
|
||||
'ai_review': {'url': '', 'timeout': ''},
|
||||
'gateway': {'url': '', 'timeout': ''},
|
||||
}
|
||||
|
||||
|
||||
@@ -93,12 +93,15 @@ if os.environ.get('DB_PORT'):
|
||||
_dnfs_config['database']['port'] = os.environ['DB_PORT']
|
||||
if os.environ.get('REVIEW_URL'):
|
||||
_dnfs_config['ai_review']['url'] = os.environ['REVIEW_URL']
|
||||
if os.environ.get('GATEWAY_URL'):
|
||||
_dnfs_config['gateway']['url'] = os.environ['GATEWAY_URL']
|
||||
|
||||
# 配置完整性校验
|
||||
_required = {
|
||||
'server': ['host', 'port'],
|
||||
'database': ['host', 'port', 'timeout'],
|
||||
'ai_review': ['url', 'timeout'],
|
||||
'gateway': ['url', 'timeout'],
|
||||
}
|
||||
for section, keys in _required.items():
|
||||
for key in keys:
|
||||
@@ -113,11 +116,14 @@ _DBV2_ADDR = _dnfs_config['database']['host']
|
||||
_DBV2_PORT = int(_dnfs_config['database']['port'])
|
||||
_REVIEW_SERVER_URL = _dnfs_config['ai_review']['url']
|
||||
_REVIEW_TIMEOUT = int(_dnfs_config['ai_review']['timeout'])
|
||||
_GATEWAY_URL = _dnfs_config['gateway']['url']
|
||||
_GATEWAY_TIMEOUT = int(_dnfs_config['gateway']['timeout'])
|
||||
_ADDR = _dnfs_config['server']['host']
|
||||
_PORT = int(_dnfs_config['server']['port'])
|
||||
|
||||
_config_source = f'confs (dnfs.ini @ {_confs_host}:{_confs_port})'
|
||||
print(f'[dnfs] 配置来源: {_config_source}')
|
||||
print(f'[dnfs] Token 验签: gateway {_GATEWAY_URL}')
|
||||
|
||||
# 管理员判定已上移至 gateway：登录时命中 [Admin] users 白名单即在 Token payload
|
||||
# 写入 is_admin=true，本服务通过 info.get('is_admin') 读取，不再本地维护名单。
|
||||
@@ -301,14 +307,46 @@ apply_security_headers(app)
|
||||
CORS(app)
|
||||
|
||||
|
||||
def decode_token(token: str):
|
||||
idx = token.find('_')
|
||||
if idx != -1:
|
||||
raw = token[:idx]
|
||||
# FIXME: 验证数字签名
|
||||
decoded = base64.b64decode(raw)
|
||||
jj = decoded.decode('utf-8')
|
||||
return json.loads(jj)
|
||||
def decode_token(token: str):
|
||||
"""通过 gateway /whoami 接口验证 Token 并返回用户信息。
|
||||
|
||||
gateway /whoami 内部完成 RSA 签名验证和过期检查，返回：
|
||||
- 成功: {"login": true, "user": "...", "is_admin": true/false, ...}
|
||||
- 失败: {"login": false}
|
||||
网络异常（gateway 不可达）按 fail-closed 处理，抛 DnfsExp 拒绝请求。
|
||||
"""
|
||||
if not token:
|
||||
return None
|
||||
|
||||
req = urllib.request.Request(
|
||||
_GATEWAY_URL,
|
||||
data=b'{}',
|
||||
headers={
|
||||
"Content-Type": "application/json",
|
||||
"Authorization": f"Bearer {token}",
|
||||
},
|
||||
method="POST",
|
||||
)
|
||||
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=_GATEWAY_TIMEOUT) as resp:
|
||||
data = json.loads(resp.read().decode("utf-8"))
|
||||
except (urllib.error.URLError, urllib.error.HTTPError, json.JSONDecodeError, OSError) as e:
|
||||
# gateway 不可达或返回异常：fail-closed，拒绝请求
|
||||
logs.warning('dnfs', f'Token 验签失败（gateway 异常）: {e}', extra={})
|
||||
raise DnfsExp('认证服务暂时不可用', 503)
|
||||
except Exception as e:
|
||||
logs.warning('dnfs', f'Token 验签未知异常: {e}', extra={})
|
||||
raise DnfsExp('认证失败', 401)
|
||||
|
||||
if not data.get('login'):
|
||||
return None
|
||||
|
||||
# gateway 返回 login=true 时 user 字段必定存在
|
||||
return {
|
||||
'user': data.get('user', ''),
|
||||
'is_admin': bool(data.get('is_admin', False)),
|
||||
}
|
||||
|
||||
|
||||
@app.route('/add_note', methods=['POST'])
|
||||
@@ -426,7 +464,7 @@ def get_notes():
|
||||
|
||||
# eg: {"id": 1}
|
||||
@app.route('/delete_note', methods=['POST'])
|
||||
|
||||
@_handle_exp
|
||||
def delete_note():
|
||||
|
||||
data = request.get_json()
|
||||
@@ -529,7 +567,7 @@ def delete_note():
|
||||
# "content": "hello"
|
||||
#}
|
||||
@app.route('/update_note', methods=['POST'])
|
||||
|
||||
@_handle_exp
|
||||
def update_note():
|
||||
|
||||
data = request.get_json()
|
||||
|
||||
@@ -83,6 +83,11 @@ services:
|
||||
- DB_HOST=db
|
||||
- DB_PORT=8005
|
||||
- REVIEW_URL=http://ai_review:8007/review
|
||||
- GATEWAY_URL=http://host.docker.internal:8443/whoami
|
||||
- CONFS_HOST=host.docker.internal
|
||||
- CONFS_PORT=8010
|
||||
extra_hosts:
|
||||
- "host.docker.internal:host-gateway"
|
||||
depends_on:
|
||||
db:
|
||||
condition: service_healthy
|
||||
|
||||
+29
-2
@@ -28,6 +28,29 @@ UniServe Framework 的应用层网关，C++20 实现。
|
||||
| CAPTCHA | 一次性、5 分钟过期，登录与注册均可选启用 |
|
||||
| 速率限制 | 内存计数器，自动清理超过 1 万条的老条目 |
|
||||
| TLS 可选 | `Server.tls = off` 时纯 HTTP（配合 WAF），`on` 时用 `full_chain.crt` / `server.key` |
|
||||
| 下游服务容错 | pymsg 客户端 `retry_on_refused` 机制：db 等强依赖服务未启动时自动阻塞重试，避免启动顺序问题；日志客户端 fail-safe 静默丢弃 |
|
||||
|
||||
## pymsg 内部通信协议
|
||||
|
||||
gateway 通过项目内的 [pymsg](../pymsg/)（长度前缀帧 TCP 协议）与内部微服务通信，C++ 实现在 `inc/Msg.hpp`，与 Python 端字节级兼容。
|
||||
|
||||
**线协议**：`[4字节大端uint32长度][UTF-8载荷]`，最大 64 MiB。
|
||||
|
||||
**JSON RPC 约定**：请求形如 `{"_route": "<路由名>", ...}`，响应形如 `{"success": true, "data": ...}` 或 `{"success": false, "error": "..."}`。
|
||||
|
||||
**下游服务调用方式**：
|
||||
|
||||
| 下游 | 端口 | `_route` | 调用位置 | 重试策略 |
|
||||
|------|------|----------|---------|---------|
|
||||
| db | 8005 | `execute` | `DbC::call_msg()`（所有数据库 CRUD） | `retry_on_refused=true`，db 未就绪时阻塞等待 |
|
||||
| db | 8005 | `execute` | `LogC::_send()`（日志异步批量推送） | `retry_on_refused=false`，fail-safe 静默丢弃 |
|
||||
| captcha | 8009 | `verify` / `image` | `verify_captcha_remote()`、`/captcha/image` 代理 | 默认不重试，连接失败 fail-closed 返回 502/false |
|
||||
| smtp | 8008 | `verify_email` / `send_email` | `verify_email_code_remote()`、`/register/send_code` | 默认不重试，连接失败 fail-closed 返回 502/false |
|
||||
|
||||
**retry_on_refused 机制**（`MsgClient::call()`，对应 Python 端 `MsgClient.call(retry_on_refused=True)`）：
|
||||
- 仅对 `ECONNREFUSED`（目标端口无服务监听）重试，超时、连接重置、业务错误立即抛出
|
||||
- 固定 1 秒间隔重试；首次失败立即打印 stderr 提示，之后每 5 秒打印一次等待日志（节制日志，避免刷屏）
|
||||
- 用于解决微服务启动顺序不确定问题：gateway 启动时 db 未就绪，首次 db 调用自动阻塞等待直到 db 上线
|
||||
|
||||
## 接口
|
||||
|
||||
@@ -265,6 +288,9 @@ gateway/
|
||||
├── src/
|
||||
│ └── main.cpp # 全部实现：PBKDF2 / PasswordValidator / RateLimiter / 路由 / handler
|
||||
├── inc/
|
||||
│ ├── Msg.hpp # pymsg C++ 实现：长度前缀帧协议、MsgClient（含 retry_on_refused 重试）、MsgException 族
|
||||
│ ├── DbC.hpp # db 服务客户端：通过 pymsg 调 dbv2 execute 路由（retry_on_refused=true）
|
||||
│ ├── LogC.hpp # 统一日志客户端：异步批量队列 + pymsg INSERT（fail-safe，不重试）
|
||||
│ ├── Conf.hpp # ConfIni：基于 boost::property_tree 的 ini 解析
|
||||
│ ├── ExpBase.hpp # 异常基类（带 source_location）
|
||||
│ └── LocalTime.hpp # 本地时间工具（getCurrent / parse）
|
||||
@@ -276,14 +302,15 @@ gateway/
|
||||
|
||||
## 版本
|
||||
|
||||
`0.6.0`（见 [src/main.cpp:31](src/main.cpp#L31)）。启动时打印：
|
||||
`0.6.1`（见 [src/main.cpp:31](src/main.cpp#L31)）。启动时打印：
|
||||
|
||||
```
|
||||
Gateway 0.6.0 started
|
||||
Gateway 0.6.1 started
|
||||
```
|
||||
|
||||
近期变更：
|
||||
|
||||
- `0.6.1`：pymsg 客户端增加 `retry_on_refused` 重试机制，与 Python 端 `pymsg.MsgClient.call(retry_on_refused=True)` 行为对齐——新增 `MsgConnectionRefusedException` 异常类（区分 ECONNREFUSED 与其他连接错误）；`MsgClient::call()` 新增 `retry_on_refused`/`retry_interval` 参数，仅对连接拒绝重试，固定间隔 1s，节制日志（首次立即打印 + 每 5s 一次），其他异常（超时/连接重置/业务错误）立即抛出；`DbC::call_msg()` 启用 `retry_on_refused=true`（db 是硬依赖，启动顺序不确定时自动阻塞等待）；`LogC::_send()` 显式传 `retry_on_refused=false`（日志是 fail-safe，db 不可用时静默丢弃，绝不阻塞后台线程）；captcha/smtp 辅助服务保持默认不重试（fail-fast，请求线程中永久阻塞会耗尽线程池）。
|
||||
- `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 内部错误细节不向外暴露）。
|
||||
|
||||
+14
-12
@@ -25,18 +25,20 @@ namespace qing {
|
||||
}
|
||||
|
||||
// 新版 pymsg 接口：通过 MsgClient（TCP 长度前缀帧协议）调用 dbv2。
|
||||
// 自动添加 _route = "execute"，返回 dbv2 pymsg 完整 JSON 响应
|
||||
// {"success":..., "data":[...]}。
|
||||
// 响应格式与 call_http 兼容，调用方可直接替换。
|
||||
static json::value call_msg(const json::value& params,
|
||||
double timeout = DEFAULT_TIMEOUT) {
|
||||
json::value jv_route = params;
|
||||
jv_route.as_object()["_route"] = "execute";
|
||||
pmc::net::MsgClient client("127.0.0.1", 8005, "gateway");
|
||||
std::string resp = client.call(
|
||||
json::serialize(jv_route), timeout);
|
||||
return json::parse(resp);
|
||||
}
|
||||
// 自动添加 _route = "execute"，返回 dbv2 pymsg 完整 JSON 响应
|
||||
// {"success":..., "data":[...]}。
|
||||
// 响应格式与 call_http 兼容，调用方可直接替换。
|
||||
// db 是强依赖，启用 retry_on_refused：db 未启动时阻塞等待，避免启动顺序问题。
|
||||
static json::value call_msg(const json::value& params,
|
||||
double timeout = DEFAULT_TIMEOUT) {
|
||||
json::value jv_route = params;
|
||||
jv_route.as_object()["_route"] = "execute";
|
||||
pmc::net::MsgClient client("127.0.0.1", 8005, "gateway");
|
||||
std::string resp = client.call(
|
||||
json::serialize(jv_route), timeout,
|
||||
/*retry_on_refused=*/true);
|
||||
return json::parse(resp);
|
||||
}
|
||||
|
||||
private:
|
||||
static constexpr const char* _DBURL = "http://127.0.0.1:8005/execute";
|
||||
|
||||
@@ -196,7 +196,8 @@ private:
|
||||
payload["values"] = std::move(values);
|
||||
|
||||
const std::string resp = client.call(
|
||||
json::serialize(payload), _TIMEOUT);
|
||||
json::serialize(payload), _TIMEOUT,
|
||||
/*retry_on_refused=*/false); // 日志是 fail-safe，db 不可用时静默丢弃，绝不阻塞重试
|
||||
auto obj = json::parse(resp).as_object();
|
||||
if (!obj.at("success").as_bool()) {
|
||||
std::string err = "unknown";
|
||||
|
||||
+75
-16
@@ -17,11 +17,14 @@
|
||||
*/
|
||||
|
||||
#include <cerrno>
|
||||
#include <chrono>
|
||||
#include <cstdint>
|
||||
#include <cstdio>
|
||||
#include <cstring>
|
||||
#include <functional>
|
||||
#include <stdexcept>
|
||||
#include <string>
|
||||
#include <thread>
|
||||
#include <vector>
|
||||
|
||||
#include <arpa/inet.h>
|
||||
@@ -59,6 +62,13 @@ public:
|
||||
explicit MsgConnectionException(const std::string &msg) : MsgException(msg) {}
|
||||
};
|
||||
|
||||
/// 目标服务未启动（connect 返回 ECONNREFUSED），用于 retry_on_refused 重试机制
|
||||
class MsgConnectionRefusedException : public MsgConnectionException {
|
||||
public:
|
||||
explicit MsgConnectionRefusedException(const std::string &msg)
|
||||
: MsgConnectionException(msg) {}
|
||||
};
|
||||
|
||||
// ============================================================================
|
||||
// Low-level protocol: 4-byte big-endian length prefix + UTF-8 payload
|
||||
// ============================================================================
|
||||
@@ -155,10 +165,14 @@ inline int openConnection(const std::string& addr, int port, double timeout) {
|
||||
|
||||
if (::connect(fd, reinterpret_cast<struct sockaddr *>(&serv_addr),
|
||||
sizeof(serv_addr)) < 0) {
|
||||
int saved_errno = errno;
|
||||
::close(fd);
|
||||
throw MsgConnectionException(
|
||||
"could not connect to " + addr + ":" + std::to_string(port) +
|
||||
" - " + ::strerror(errno));
|
||||
std::string err_msg = "could not connect to " + addr + ":" +
|
||||
std::to_string(port) + " - " + ::strerror(saved_errno);
|
||||
if (saved_errno == ECONNREFUSED) {
|
||||
throw MsgConnectionRefusedException(err_msg);
|
||||
}
|
||||
throw MsgConnectionException(err_msg);
|
||||
}
|
||||
|
||||
return fd;
|
||||
@@ -242,19 +256,64 @@ public:
|
||||
///
|
||||
/// 不会给消息添加 ``name:`` 前缀（适用于 RPC/API 调用）。
|
||||
///
|
||||
/// @param msg 要发送的原始字符串。
|
||||
/// @param timeout 超时秒数，默认 10.0。
|
||||
/// @returns 服务器返回的字符串。
|
||||
std::string call(const std::string &msg, double timeout = 10.0) {
|
||||
int fd = detail::openConnection(_addr, _port, timeout);
|
||||
try {
|
||||
pmsg_send(fd, msg);
|
||||
std::string resp = pmsg_recv(fd);
|
||||
::close(fd);
|
||||
return resp;
|
||||
} catch (...) {
|
||||
::close(fd);
|
||||
throw;
|
||||
/// @param msg 要发送的原始字符串。
|
||||
/// @param timeout 超时秒数，默认 10.0。
|
||||
/// @param retry_on_refused 目标服务未启动（ECONNREFUSED）时是否阻塞重试，
|
||||
/// 直到连接成功。适用于启动顺序不确定的服务发现场景。
|
||||
/// 默认 false（连接拒绝立即抛异常）。
|
||||
/// @param retry_interval 重试间隔秒数，默认 1.0。
|
||||
/// @returns 服务器返回的字符串。
|
||||
std::string call(const std::string &msg, double timeout = 10.0,
|
||||
bool retry_on_refused = false,
|
||||
double retry_interval = 1.0) {
|
||||
if (!retry_on_refused) {
|
||||
// 快速路径：不重试，直接执行一次，保持原有行为
|
||||
int fd = detail::openConnection(_addr, _port, timeout);
|
||||
try {
|
||||
pmsg_send(fd, msg);
|
||||
std::string resp = pmsg_recv(fd);
|
||||
::close(fd);
|
||||
return resp;
|
||||
} catch (...) {
|
||||
::close(fd);
|
||||
throw;
|
||||
}
|
||||
}
|
||||
|
||||
// 重试路径：仅对 ECONNREFUSED 重试，其他错误立即抛出
|
||||
constexpr double kLogEvery = 5.0; // 重试期间打印等待提示的间隔（秒）
|
||||
int attempt = 0;
|
||||
auto last_log = std::chrono::steady_clock::now() -
|
||||
std::chrono::duration_cast<std::chrono::steady_clock::duration>(
|
||||
std::chrono::duration<double>(kLogEvery));
|
||||
|
||||
while (true) {
|
||||
try {
|
||||
int fd = detail::openConnection(_addr, _port, timeout);
|
||||
try {
|
||||
pmsg_send(fd, msg);
|
||||
std::string resp = pmsg_recv(fd);
|
||||
::close(fd);
|
||||
return resp;
|
||||
} catch (...) {
|
||||
::close(fd);
|
||||
throw;
|
||||
}
|
||||
} catch (const MsgConnectionRefusedException &) {
|
||||
++attempt;
|
||||
auto now = std::chrono::steady_clock::now();
|
||||
double elapsed = std::chrono::duration<double>(now - last_log).count();
|
||||
if (elapsed >= kLogEvery || attempt == 1) {
|
||||
std::fprintf(stderr,
|
||||
"[pymsg] 等待服务 %s:%d 就绪... (已重试 %d 次)\n",
|
||||
_addr.c_str(), _port, attempt);
|
||||
std::fflush(stderr);
|
||||
last_log = now;
|
||||
}
|
||||
std::this_thread::sleep_for(
|
||||
std::chrono::duration<double>(retry_interval));
|
||||
}
|
||||
// 其他异常（MsgTimeoutException、普通MsgConnectionException等）不捕获，直接向上抛出
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -28,7 +28,7 @@ using namespace qing;
|
||||
namespace json = boost::json;
|
||||
namespace po = boost::program_options;
|
||||
|
||||
constexpr const char *_VER = "0.6.0";
|
||||
constexpr const char *_VER = "0.6.1";
|
||||
|
||||
//////////////////////////////////
|
||||
#include <sstream>
|
||||
|
||||
+35
-3
@@ -13,15 +13,38 @@ CLIENT_LIBS := -L/opt/homebrew/lib -lssl -lcrypto -lcurl -lpthread
|
||||
BUILD_TDIR := build
|
||||
$(shell mkdir -p $(BUILD_TDIR))
|
||||
|
||||
# Server object files (HttpServer 拆分后)
|
||||
SERVER_OBJS := $(BUILD_TDIR)/HttpServer.o \
|
||||
$(BUILD_TDIR)/HttpServer_Listener.o \
|
||||
$(BUILD_TDIR)/HttpServer_Routes.o \
|
||||
$(BUILD_TDIR)/HttpServer_Request.o \
|
||||
$(BUILD_TDIR)/HttpServer_Ws.o \
|
||||
$(BUILD_TDIR)/HttpServer_WsSession.o
|
||||
|
||||
#--------
|
||||
all: $(BUILD_TDIR)/$(LIB_NAME).so $(BUILD_TDIR)/$(LIBC_NAME).so
|
||||
|
||||
# ---- Server library (HttpServer, no curl) ----
|
||||
$(BUILD_TDIR)/$(LIB_NAME).so: $(BUILD_TDIR)/HttpServer.o
|
||||
$(BUILD_TDIR)/$(LIB_NAME).so: $(SERVER_OBJS)
|
||||
$(CPP) -shared $^ $(SERVER_LIBS) -o $@
|
||||
|
||||
$(BUILD_TDIR)/HttpServer.o: src/HttpServer.cpp
|
||||
$(CPP) $(CPPFLAGS) -fPIC -c $^ $(INCS) -o $@
|
||||
$(BUILD_TDIR)/HttpServer.o: src/HttpServer.cpp inc/us/detail/HttpServerImpl.hpp
|
||||
$(CPP) $(CPPFLAGS) -fPIC -c $< $(INCS) -o $@
|
||||
|
||||
$(BUILD_TDIR)/HttpServer_Listener.o: src/HttpServer_Listener.cpp inc/us/detail/HttpServerImpl.hpp
|
||||
$(CPP) $(CPPFLAGS) -fPIC -c $< $(INCS) -o $@
|
||||
|
||||
$(BUILD_TDIR)/HttpServer_Routes.o: src/HttpServer_Routes.cpp inc/us/detail/HttpServerImpl.hpp
|
||||
$(CPP) $(CPPFLAGS) -fPIC -c $< $(INCS) -o $@
|
||||
|
||||
$(BUILD_TDIR)/HttpServer_Request.o: src/HttpServer_Request.cpp inc/us/detail/HttpServerImpl.hpp
|
||||
$(CPP) $(CPPFLAGS) -fPIC -c $< $(INCS) -o $@
|
||||
|
||||
$(BUILD_TDIR)/HttpServer_Ws.o: src/HttpServer_Ws.cpp inc/us/detail/HttpServerImpl.hpp
|
||||
$(CPP) $(CPPFLAGS) -fPIC -c $< $(INCS) -o $@
|
||||
|
||||
$(BUILD_TDIR)/HttpServer_WsSession.o: src/HttpServer_WsSession.cpp inc/us/detail/HttpServerImpl.hpp
|
||||
$(CPP) $(CPPFLAGS) -fPIC -c $< $(INCS) -o $@
|
||||
|
||||
# ---- Client library (HttpClient, depends on curl + beast websocket) ----
|
||||
$(BUILD_TDIR)/$(LIBC_NAME).so: $(BUILD_TDIR)/HttpClient.o
|
||||
@@ -36,8 +59,13 @@ clear:
|
||||
install: all
|
||||
@echo "install UniServe_libhttp..."
|
||||
@mkdir -vp $(INCLU_TDIR)/$(FRAMEWORK_NAME)
|
||||
@mkdir -vp $(INCLU_TDIR)/$(FRAMEWORK_NAME)/detail
|
||||
@cp -vf $(INCLU_SDIR)/$(FRAMEWORK_NAME)/HttpServer.hpp $(INCLU_TDIR)/$(FRAMEWORK_NAME)/
|
||||
@cp -vf $(INCLU_SDIR)/$(FRAMEWORK_NAME)/HttpClient.hpp $(INCLU_TDIR)/$(FRAMEWORK_NAME)/
|
||||
@cp -vf $(INCLU_SDIR)/$(FRAMEWORK_NAME)/WebSocketServer.hpp $(INCLU_TDIR)/$(FRAMEWORK_NAME)/
|
||||
@cp -vf $(INCLU_SDIR)/$(FRAMEWORK_NAME)/WebSocketClient.hpp $(INCLU_TDIR)/$(FRAMEWORK_NAME)/
|
||||
@cp -vf $(INCLU_SDIR)/$(FRAMEWORK_NAME)/WebSocketClientBase.hpp $(INCLU_TDIR)/$(FRAMEWORK_NAME)/
|
||||
@cp -vf $(INCLU_SDIR)/$(FRAMEWORK_NAME)/detail/HttpServerImpl.hpp $(INCLU_TDIR)/$(FRAMEWORK_NAME)/detail/
|
||||
@cp -vf $(BUILD_TDIR)/$(LIB_NAME).so $(INSTL_TDIR)/
|
||||
@cp -vf $(BUILD_TDIR)/$(LIBC_NAME).so $(INSTL_TDIR)/
|
||||
@echo "done."
|
||||
@@ -46,6 +74,10 @@ uninstall:
|
||||
@echo "uninstall UniServe_libhttp..."
|
||||
@rm -vf $(INCLU_TDIR)/$(FRAMEWORK_NAME)/HttpServer.hpp
|
||||
@rm -vf $(INCLU_TDIR)/$(FRAMEWORK_NAME)/HttpClient.hpp
|
||||
@rm -vf $(INCLU_TDIR)/$(FRAMEWORK_NAME)/WebSocketServer.hpp
|
||||
@rm -vf $(INCLU_TDIR)/$(FRAMEWORK_NAME)/WebSocketClient.hpp
|
||||
@rm -vf $(INCLU_TDIR)/$(FRAMEWORK_NAME)/WebSocketClientBase.hpp
|
||||
@rm -vf $(INCLU_TDIR)/$(FRAMEWORK_NAME)/detail/HttpServerImpl.hpp
|
||||
@rm -vf $(INSTL_TDIR)/$(LIB_NAME).so
|
||||
@rm -vf $(INSTL_TDIR)/$(LIBC_NAME).so
|
||||
@echo "done."
|
||||
@@ -46,6 +46,23 @@ Two shared libraries are produced:
|
||||
|
||||
`libus_http.so` does **not** depend on libcurl. Link against `libus_httpclient.so` only when you need the client, and link both when you need both server and client (e.g. a proxy/gateway).
|
||||
|
||||
### Source Layout
|
||||
|
||||
The server implementation is split into modular source files for faster incremental builds:
|
||||
|
||||
| Source File | Responsibility |
|
||||
|---|---|
|
||||
| `src/HttpServer.cpp` | Core lifecycle (construct/destruct/start/stop/run), SSL init, session management, debug utilities |
|
||||
| `src/HttpServer_Listener.cpp` | TCP connection acceptor (`Listener` class) |
|
||||
| `src/HttpServer_Routes.cpp` | HTTP route registration, route matching, wildcard→regex conversion, query parameter parsing |
|
||||
| `src/HttpServer_Request.cpp` | Request dispatch pipeline, static file serving, CORS preflight, Session HTTP read/write |
|
||||
| `src/HttpServer_Ws.cpp` | WebSocket route registration, client registry, broadcast/send APIs |
|
||||
| `src/HttpServer_WsSession.cpp` | Session WebSocket handling (regular WS frames + bidirectional WS proxy bridging) |
|
||||
|
||||
Internal shared definitions (exception classes, full `Session`/`Listener` declarations) live in `inc/us/detail/HttpServerImpl.hpp` (installed to `/usr/include/us/detail/`).
|
||||
|
||||
Modifying one module only requires recompiling its corresponding `.o` file — no need to rebuild the entire 1400-line translation unit.
|
||||
|
||||
## Quick Start
|
||||
|
||||
### HTTP + WebSocket Server
|
||||
@@ -260,6 +277,7 @@ using WebSocketClientBase = HttpClient;
|
||||
|
||||
## Version History
|
||||
|
||||
- **v3.3.0**: Split HttpServer into 6 modular translation units by functional area (core/lifecycle, listener, routes, request dispatch, WS management, WS session/proxy) for dramatically faster incremental builds. Added internal detail header `inc/us/detail/HttpServerImpl.hpp` for shared definitions across TU boundaries.
|
||||
- **v3.2.0**: Split into two shared libraries — `libus_http.so` (server only, no curl) and `libus_httpclient.so` (HttpClient, depends on curl). HttpClient implementation moved to `src/HttpClient.cpp`; `HttpClient.hpp` is declaration-only.
|
||||
- **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.
|
||||
|
||||
@@ -0,0 +1,180 @@
|
||||
#ifndef HTTPSERVER_IMPL_HPP
|
||||
#define HTTPSERVER_IMPL_HPP
|
||||
|
||||
#include "us/HttpServer.hpp"
|
||||
#include <boost/algorithm/string.hpp>
|
||||
#include <fstream>
|
||||
#include <sstream>
|
||||
#include <iomanip>
|
||||
#include <openssl/err.h>
|
||||
#include <openssl/ssl.h>
|
||||
#include <boost/beast/websocket/ssl.hpp>
|
||||
|
||||
namespace pmc {
|
||||
namespace net {
|
||||
|
||||
// ===================== 自定义异常 =====================
|
||||
class HttpServerException : public std::runtime_error {
|
||||
public:
|
||||
explicit HttpServerException(const std::string& msg) : runtime_error(msg) {}
|
||||
};
|
||||
class HttpListenerException : public HttpServerException {
|
||||
public:
|
||||
explicit HttpListenerException(const std::string& msg) : HttpServerException(msg) {}
|
||||
};
|
||||
class SSLInitException : public HttpServerException {
|
||||
public:
|
||||
explicit SSLInitException(const std::string& msg) : HttpServerException(msg) {}
|
||||
};
|
||||
class CreateObjectFailed : public HttpServerException {
|
||||
public:
|
||||
explicit CreateObjectFailed(const std::string& msg) : HttpServerException(msg) {}
|
||||
};
|
||||
|
||||
// ===================== Session (完整定义) =====================
|
||||
class HttpServer::Session : public std::enable_shared_from_this<Session> {
|
||||
using tcp_stream = beast::tcp_stream;
|
||||
using ssl_stream = asio::ssl::stream<tcp_stream&>;
|
||||
using ws_stream = websocket::stream<tcp_stream&>;
|
||||
using wss_stream = websocket::stream<ssl_stream&>;
|
||||
|
||||
// Proxy outbound types
|
||||
using proxy_ws_stream = websocket::stream<tcp::socket&>;
|
||||
using proxy_wss_stream = websocket::stream<asio::ssl::stream<tcp::socket&>&>;
|
||||
|
||||
public:
|
||||
Session(tcp::socket&& socket, HttpServer* server)
|
||||
: stream_(std::move(socket)), server_(server), ssl_enabled_(false),
|
||||
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), is_ws_proxy_(false), client_id_("") {
|
||||
ssl_stream_ = std::make_unique<ssl_stream>(stream_, ssl_ctx);
|
||||
}
|
||||
|
||||
~Session() { close(); }
|
||||
|
||||
void run();
|
||||
void setRemoteInfo(const std::string& ip, unsigned short port);
|
||||
void wsSend(const std::string& message);
|
||||
std::string getClientId() const { return client_id_; }
|
||||
std::string getRemoteAddress() const { return remote_ip_ + ":" + std::to_string(remote_port_); }
|
||||
bool isWebSocket() const { return is_websocket_; }
|
||||
|
||||
private:
|
||||
void close();
|
||||
void doSSLHandshake();
|
||||
void doRead();
|
||||
void handleRead(beast::error_code ec);
|
||||
bool isWebSocketUpgrade(const http::request<http::string_body>& req);
|
||||
void doWsUpgrade();
|
||||
void doWssUpgrade();
|
||||
void onWsAccept(beast::error_code ec);
|
||||
|
||||
// Regular WebSocket
|
||||
void doWsRead();
|
||||
void onWsRead(beast::error_code ec, std::size_t bytes_transferred);
|
||||
void doWsWrite();
|
||||
void onWsClose();
|
||||
|
||||
// WebSocket Proxy
|
||||
static void parseProxyUrl(const std::string& url, bool& ssl, std::string& host,
|
||||
std::string& port, std::string& path);
|
||||
void doProxyConnect(const std::string& target_url);
|
||||
void onProxyResolve(beast::error_code ec, tcp::resolver::results_type results);
|
||||
void onProxyConnect(beast::error_code ec, tcp::endpoint ep);
|
||||
void onProxySslHandshake(beast::error_code ec);
|
||||
void onProxyHandshake(beast::error_code ec);
|
||||
|
||||
// Client → Backend
|
||||
void doProxyReadClient();
|
||||
void onProxyReadClient(beast::error_code ec, std::size_t bytes_transferred);
|
||||
void doProxyWriteBackend();
|
||||
|
||||
// Backend → Client
|
||||
void doProxyReadBackend();
|
||||
void onProxyReadBackend(beast::error_code ec, std::size_t bytes_transferred);
|
||||
void doProxyWriteClient();
|
||||
|
||||
void closeProxy(websocket::close_code code);
|
||||
|
||||
// Regular HTTP
|
||||
void processRequest(http::request<http::string_body> req);
|
||||
void doWrite(std::shared_ptr<http::response<http::string_body>> response);
|
||||
|
||||
// Client stream (inbound)
|
||||
tcp_stream stream_;
|
||||
std::unique_ptr<ssl_stream> ssl_stream_;
|
||||
std::unique_ptr<ws_stream> ws_stream_;
|
||||
std::unique_ptr<wss_stream> wss_stream_;
|
||||
|
||||
beast::flat_buffer buffer_;
|
||||
beast::flat_buffer ws_buffer_;
|
||||
http::request<http::string_body> request_;
|
||||
|
||||
HttpServer* server_;
|
||||
bool ssl_enabled_;
|
||||
|
||||
std::string remote_ip_ = "unknown";
|
||||
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<std::string> 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<tcp::resolver> proxy_resolver_;
|
||||
std::unique_ptr<tcp::socket> proxy_socket_;
|
||||
std::unique_ptr<beast::tcp_stream> proxy_tcp_stream_;
|
||||
std::unique_ptr<asio::ssl::context> proxy_ssl_ctx_;
|
||||
std::unique_ptr<asio::ssl::stream<tcp::socket&>> proxy_ssl_stream_;
|
||||
std::unique_ptr<proxy_ws_stream> proxy_ws_stream_;
|
||||
std::unique_ptr<proxy_wss_stream> proxy_wss_stream_;
|
||||
|
||||
beast::flat_buffer proxy_client_buffer_;
|
||||
beast::flat_buffer proxy_backend_buffer_;
|
||||
std::deque<std::pair<std::string, bool>> proxy_to_backend_queue_;
|
||||
std::deque<std::pair<std::string, bool>> proxy_to_client_queue_;
|
||||
};
|
||||
|
||||
// ===================== Listener (完整定义) =====================
|
||||
class HttpServer::Listener : public std::enable_shared_from_this<Listener> {
|
||||
public:
|
||||
Listener(asio::io_context& ioc, tcp::endpoint endpoint, HttpServer* server)
|
||||
: ioc_(ioc), acceptor_(ioc), server_(server), ssl_enabled_(false) {
|
||||
initAcceptor(endpoint);
|
||||
}
|
||||
|
||||
Listener(asio::io_context& ioc, tcp::endpoint endpoint,
|
||||
asio::ssl::context& ssl_ctx, HttpServer* server)
|
||||
: ioc_(ioc), acceptor_(ioc), ssl_ctx_(&ssl_ctx), server_(server), ssl_enabled_(true) {
|
||||
initAcceptor(endpoint);
|
||||
}
|
||||
|
||||
void run();
|
||||
|
||||
private:
|
||||
void initAcceptor(tcp::endpoint ep);
|
||||
void doAccept();
|
||||
|
||||
asio::io_context& ioc_;
|
||||
tcp::acceptor acceptor_;
|
||||
asio::ssl::context* ssl_ctx_ = nullptr;
|
||||
HttpServer* server_;
|
||||
bool ssl_enabled_;
|
||||
};
|
||||
|
||||
} // namespace net
|
||||
} // namespace pmc
|
||||
|
||||
#endif // HTTPSERVER_IMPL_HPP
|
||||
+23
-1224
File diff suppressed because it is too large.
Load diff
@@ -0,0 +1,51 @@
|
||||
#include "us/detail/HttpServerImpl.hpp"
|
||||
|
||||
namespace pmc {
|
||||
namespace net {
|
||||
|
||||
// ===================== Listener 实现 =====================
|
||||
|
||||
void HttpServer::Listener::run() {
|
||||
if (acceptor_.is_open()) doAccept();
|
||||
}
|
||||
|
||||
void HttpServer::Listener::initAcceptor(tcp::endpoint ep) {
|
||||
beast::error_code ec;
|
||||
acceptor_.open(ep.protocol(), ec);
|
||||
if (!ec) {
|
||||
acceptor_.set_option(asio::socket_base::reuse_address(true), ec);
|
||||
acceptor_.bind(ep, ec);
|
||||
if (!ec) acceptor_.listen(asio::socket_base::max_listen_connections, ec);
|
||||
}
|
||||
if (ec) throw HttpListenerException("Listener error: " + ec.message());
|
||||
}
|
||||
|
||||
void HttpServer::Listener::doAccept() {
|
||||
auto self = shared_from_this();
|
||||
acceptor_.async_accept(
|
||||
asio::make_strand(ioc_),
|
||||
[self](beast::error_code ec, tcp::socket socket) {
|
||||
if (!ec) {
|
||||
auto ep = socket.remote_endpoint(ec);
|
||||
std::string ip = ec ? "unknown" : ep.address().to_string();
|
||||
unsigned short port = ec ? 0 : ep.port();
|
||||
std::cout << "[Listener] New connection from " << ip << ":" << port << std::endl;
|
||||
|
||||
std::shared_ptr<Session> session;
|
||||
if (self->ssl_enabled_ && self->ssl_ctx_) {
|
||||
session = std::make_shared<Session>(std::move(socket), *self->ssl_ctx_, self->server_);
|
||||
} else {
|
||||
session = std::make_shared<Session>(std::move(socket), self->server_);
|
||||
}
|
||||
session->setRemoteInfo(ip, port);
|
||||
self->server_->addSession(session);
|
||||
session->run();
|
||||
} else {
|
||||
std::cerr << "[Listener] Accept error: " << ec.message() << std::endl;
|
||||
}
|
||||
self->doAccept();
|
||||
});
|
||||
}
|
||||
|
||||
} // namespace net
|
||||
} // namespace pmc
|
||||
@@ -0,0 +1,329 @@
|
||||
#include "us/detail/HttpServerImpl.hpp"
|
||||
|
||||
namespace pmc {
|
||||
namespace net {
|
||||
|
||||
// ---- 请求处理核心 ----
|
||||
http::response<http::string_body> HttpServer::handleRequest(const http::request<http::string_body>& req) {
|
||||
std::string target_path(req.target().data(), req.target().size());
|
||||
std::string query_string;
|
||||
auto pos = target_path.find('?');
|
||||
if (pos != std::string::npos) {
|
||||
query_string = target_path.substr(pos + 1);
|
||||
target_path = target_path.substr(0, pos);
|
||||
}
|
||||
auto params = parseQueryParams(query_string);
|
||||
|
||||
http::response<http::string_body> res;
|
||||
res.version(req.version());
|
||||
res.set(http::field::server, "Boost.Beast HttpServer");
|
||||
res.set(http::field::content_type, "text/plain");
|
||||
res.keep_alive(req.keep_alive());
|
||||
|
||||
bool middleware_handled = false;
|
||||
for (auto& m : middlewares_) {
|
||||
if (!m(req, res, params)) {
|
||||
middleware_handled = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
if (middleware_handled) {
|
||||
res.prepare_payload();
|
||||
return res;
|
||||
}
|
||||
|
||||
std::string method = http::to_string(req.method());
|
||||
|
||||
if (method == "OPTIONS") {
|
||||
auto handler = findRoute("OPTIONS", target_path);
|
||||
if (handler) {
|
||||
try {
|
||||
return handler(req, params);
|
||||
} catch (const std::exception& e) {
|
||||
std::cerr << "[Handler] Exception: " << e.what() << std::endl;
|
||||
res.result(http::status::internal_server_error);
|
||||
res.body() = "500 Internal Server Error\n";
|
||||
res.prepare_payload();
|
||||
return res;
|
||||
}
|
||||
}
|
||||
|
||||
auto origin = req.find(http::field::origin);
|
||||
if (origin != req.end()) {
|
||||
std::vector<std::string> all_methods = {"GET", "POST", "PUT", "DELETE", "OPTIONS"};
|
||||
std::vector<std::string> allowed_methods;
|
||||
|
||||
for (const auto& m : all_methods) {
|
||||
if (m == "OPTIONS" || routeExists(m, target_path)) {
|
||||
allowed_methods.push_back(m);
|
||||
}
|
||||
}
|
||||
|
||||
if (!allowed_methods.empty() || findWsRoute(target_path)) {
|
||||
if (allowed_methods.empty()) allowed_methods.push_back("GET");
|
||||
|
||||
std::string allow_methods;
|
||||
for (size_t i = 0; i < allowed_methods.size(); ++i) {
|
||||
if (i > 0) allow_methods += ", ";
|
||||
allow_methods += allowed_methods[i];
|
||||
}
|
||||
|
||||
res.result(http::status::ok);
|
||||
res.set(http::field::allow, allow_methods);
|
||||
res.set(http::field::access_control_allow_origin,
|
||||
std::string(origin->value()));
|
||||
res.set(http::field::access_control_allow_methods, allow_methods);
|
||||
res.set(http::field::access_control_allow_headers,
|
||||
"Content-Type, Authorization, X-Requested-With, Accept, Origin");
|
||||
res.set(http::field::access_control_max_age, "86400");
|
||||
res.body() = "";
|
||||
res.content_length(0);
|
||||
res.prepare_payload();
|
||||
return res;
|
||||
}
|
||||
}
|
||||
|
||||
res.result(http::status::not_found);
|
||||
res.body() = "404 Not Found\n";
|
||||
res.prepare_payload();
|
||||
return res;
|
||||
}
|
||||
|
||||
auto handler = findRoute(method, target_path);
|
||||
|
||||
if (!handler && !static_directory_.empty()) {
|
||||
std::string file_path = static_directory_ + target_path;
|
||||
std::ifstream file(file_path, std::ios::binary);
|
||||
if (file.is_open()) {
|
||||
std::stringstream buffer;
|
||||
buffer << file.rdbuf();
|
||||
res.result(http::status::ok);
|
||||
res.body() = buffer.str();
|
||||
|
||||
if (boost::algorithm::ends_with(file_path, ".html"))
|
||||
res.set(http::field::content_type, "text/html");
|
||||
else if (boost::algorithm::ends_with(file_path, ".css"))
|
||||
res.set(http::field::content_type, "text/css");
|
||||
else if (boost::algorithm::ends_with(file_path, ".js"))
|
||||
res.set(http::field::content_type, "application/javascript");
|
||||
else if (boost::algorithm::ends_with(file_path, ".json"))
|
||||
res.set(http::field::content_type, "application/json");
|
||||
else if (boost::algorithm::ends_with(file_path, ".png"))
|
||||
res.set(http::field::content_type, "image/png");
|
||||
else if (boost::algorithm::ends_with(file_path, ".jpg") ||
|
||||
boost::algorithm::ends_with(file_path, ".jpeg"))
|
||||
res.set(http::field::content_type, "image/jpeg");
|
||||
res.prepare_payload();
|
||||
return res;
|
||||
}
|
||||
}
|
||||
|
||||
if (handler) {
|
||||
try {
|
||||
return handler(req, params);
|
||||
} catch (const std::exception& e) {
|
||||
std::cerr << "[Handler] Exception: " << e.what() << std::endl;
|
||||
res.result(http::status::internal_server_error);
|
||||
res.body() = "500 Internal Server Error\n";
|
||||
res.prepare_payload();
|
||||
return res;
|
||||
}
|
||||
}
|
||||
|
||||
res.result(http::status::not_found);
|
||||
res.body() = "404 Not Found\n";
|
||||
res.prepare_payload();
|
||||
return res;
|
||||
}
|
||||
|
||||
// ---- Session HTTP 处理 ----
|
||||
void HttpServer::Session::setRemoteInfo(const std::string& ip, unsigned short port) {
|
||||
remote_ip_ = ip;
|
||||
remote_port_ = port;
|
||||
}
|
||||
|
||||
void HttpServer::Session::doSSLHandshake() {
|
||||
auto self = shared_from_this();
|
||||
std::cout << "[SSL] Starting handshake for "
|
||||
<< remote_ip_ << ":" << remote_port_ << std::endl;
|
||||
|
||||
ssl_stream_->async_handshake(
|
||||
asio::ssl::stream_base::server,
|
||||
[self](beast::error_code ec) {
|
||||
if (ec) {
|
||||
std::cerr << "[SSL] Handshake error: "
|
||||
<< self->remote_ip_ << ":" << self->remote_port_
|
||||
<< " - " << ec.message()
|
||||
<< " (code=" << ec.value() << ")" << std::endl;
|
||||
ERR_print_errors_fp(stderr);
|
||||
return;
|
||||
}
|
||||
std::cout << "[SSL] Handshake succeeded: "
|
||||
<< self->remote_ip_ << ":" << self->remote_port_ << std::endl;
|
||||
self->doRead();
|
||||
});
|
||||
}
|
||||
|
||||
void HttpServer::Session::doRead() {
|
||||
auto self = shared_from_this();
|
||||
request_ = {};
|
||||
buffer_.consume(buffer_.size());
|
||||
|
||||
if (ssl_enabled_ && ssl_stream_) {
|
||||
http::async_read(*ssl_stream_, buffer_, request_,
|
||||
[self](beast::error_code ec, std::size_t) {
|
||||
self->handleRead(ec);
|
||||
});
|
||||
} else {
|
||||
http::async_read(stream_, buffer_, request_,
|
||||
[self](beast::error_code ec, std::size_t) {
|
||||
self->handleRead(ec);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
bool HttpServer::Session::isWebSocketUpgrade(const http::request<http::string_body>& req) {
|
||||
if (req.method() != http::verb::get) return false;
|
||||
|
||||
auto upgrade = req.find(http::field::upgrade);
|
||||
if (upgrade == req.end()) return false;
|
||||
|
||||
std::string upgrade_val(upgrade->value());
|
||||
boost::to_lower(upgrade_val);
|
||||
boost::trim(upgrade_val);
|
||||
if (upgrade_val != "websocket") return false;
|
||||
|
||||
auto connection = req.find(http::field::connection);
|
||||
if (connection == req.end()) return false;
|
||||
|
||||
std::string conn_val(connection->value());
|
||||
boost::to_lower(conn_val);
|
||||
return conn_val.find("upgrade") != std::string::npos;
|
||||
}
|
||||
|
||||
void HttpServer::Session::handleRead(beast::error_code ec) {
|
||||
if (!ec) {
|
||||
if (isWebSocketUpgrade(request_)) {
|
||||
std::string path = std::string(request_.target());
|
||||
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<http::response<http::string_body>>();
|
||||
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;
|
||||
ws_path_ = path;
|
||||
if (ssl_enabled_) {
|
||||
doWssUpgrade();
|
||||
} else {
|
||||
doWsUpgrade();
|
||||
}
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
auto req_copy = std::move(request_);
|
||||
processRequest(std::move(req_copy));
|
||||
} else if (ec != http::error::end_of_stream) {
|
||||
std::cerr << "[Read] Error from " << remote_ip_ << ":" << remote_port_
|
||||
<< ": " << ec.message() << std::endl;
|
||||
}
|
||||
}
|
||||
|
||||
void HttpServer::Session::processRequest(http::request<http::string_body> req) {
|
||||
try {
|
||||
auto resp = std::make_shared<http::response<http::string_body>>(
|
||||
server_->handleRequest(req));
|
||||
doWrite(resp);
|
||||
} catch (const std::exception& e) {
|
||||
std::cerr << "[Request] Handler exception: " << e.what() << std::endl;
|
||||
auto err_resp = std::make_shared<http::response<http::string_body>>();
|
||||
err_resp->version(req.version());
|
||||
err_resp->result(http::status::internal_server_error);
|
||||
err_resp->set(http::field::content_type, "text/plain");
|
||||
err_resp->body() = "500 Internal Server Error\n";
|
||||
err_resp->prepare_payload();
|
||||
doWrite(err_resp);
|
||||
}
|
||||
}
|
||||
|
||||
void HttpServer::Session::doWrite(std::shared_ptr<http::response<http::string_body>> response) {
|
||||
auto self = shared_from_this();
|
||||
if (ssl_enabled_ && ssl_stream_) {
|
||||
http::async_write(*ssl_stream_, *response,
|
||||
[self, response](beast::error_code ec, std::size_t) {
|
||||
if (ec) {
|
||||
std::cerr << "[Write] Error to " << self->remote_ip_ << ":" << self->remote_port_
|
||||
<< ": " << ec.message() << std::endl;
|
||||
return;
|
||||
}
|
||||
if (!response->keep_alive()) self->close();
|
||||
else self->doRead();
|
||||
});
|
||||
} else {
|
||||
http::async_write(stream_, *response,
|
||||
[self, response](beast::error_code ec, std::size_t) {
|
||||
if (ec) {
|
||||
std::cerr << "[Write] Error to " << self->remote_ip_ << ":" << self->remote_port_
|
||||
<< ": " << ec.message() << std::endl;
|
||||
return;
|
||||
}
|
||||
if (!response->keep_alive()) self->close();
|
||||
else self->doRead();
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
void HttpServer::Session::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);
|
||||
} else if (ws_stream_) {
|
||||
ws_stream_->close(websocket::close_code::normal, ec);
|
||||
}
|
||||
}
|
||||
if (ssl_stream_) {
|
||||
ssl_stream_->shutdown(ec);
|
||||
}
|
||||
stream_.socket().shutdown(tcp::socket::shutdown_send, ec);
|
||||
stream_.close();
|
||||
}
|
||||
|
||||
void HttpServer::Session::run() {
|
||||
if (ssl_enabled_) {
|
||||
doSSLHandshake();
|
||||
} else {
|
||||
doRead();
|
||||
}
|
||||
}
|
||||
|
||||
} // namespace net
|
||||
} // namespace pmc
|
||||
@@ -0,0 +1,125 @@
|
||||
#include "us/detail/HttpServerImpl.hpp"
|
||||
|
||||
namespace pmc {
|
||||
namespace net {
|
||||
|
||||
// ---- HTTP 路由注册 ----
|
||||
void HttpServer::get(const std::string& path, HttpRequestHandler handler) {
|
||||
RouteEntry entry;
|
||||
entry.path = path;
|
||||
entry.handler = handler;
|
||||
entry.is_wildcard = (path.find('*') != std::string::npos ||
|
||||
path.find('?') != std::string::npos);
|
||||
if (entry.is_wildcard) entry.pattern = wildcardToRegex(path);
|
||||
get_handlers_.push_back(entry);
|
||||
}
|
||||
|
||||
void HttpServer::post(const std::string& path, HttpRequestHandler handler) {
|
||||
RouteEntry entry;
|
||||
entry.path = path;
|
||||
entry.handler = handler;
|
||||
entry.is_wildcard = (path.find('*') != std::string::npos ||
|
||||
path.find('?') != std::string::npos);
|
||||
if (entry.is_wildcard) entry.pattern = wildcardToRegex(path);
|
||||
post_handlers_.push_back(entry);
|
||||
}
|
||||
|
||||
void HttpServer::put(const std::string& path, HttpRequestHandler handler) {
|
||||
RouteEntry entry;
|
||||
entry.path = path;
|
||||
entry.handler = handler;
|
||||
entry.is_wildcard = (path.find('*') != std::string::npos ||
|
||||
path.find('?') != std::string::npos);
|
||||
if (entry.is_wildcard) entry.pattern = wildcardToRegex(path);
|
||||
put_handlers_.push_back(entry);
|
||||
}
|
||||
|
||||
void HttpServer::del(const std::string& path, HttpRequestHandler handler) {
|
||||
RouteEntry entry;
|
||||
entry.path = path;
|
||||
entry.handler = handler;
|
||||
entry.is_wildcard = (path.find('*') != std::string::npos ||
|
||||
path.find('?') != std::string::npos);
|
||||
if (entry.is_wildcard) entry.pattern = wildcardToRegex(path);
|
||||
delete_handlers_.push_back(entry);
|
||||
}
|
||||
|
||||
void HttpServer::options(const std::string& path, HttpRequestHandler handler) {
|
||||
RouteEntry entry;
|
||||
entry.path = path;
|
||||
entry.handler = handler;
|
||||
entry.is_wildcard = (path.find('*') != std::string::npos ||
|
||||
path.find('?') != std::string::npos);
|
||||
if (entry.is_wildcard) entry.pattern = wildcardToRegex(path);
|
||||
options_handlers_.push_back(entry);
|
||||
}
|
||||
|
||||
void HttpServer::use(MiddlewareHandler m) {
|
||||
middlewares_.push_back(m);
|
||||
}
|
||||
|
||||
void HttpServer::setStaticDirectory(const std::string& p) {
|
||||
static_directory_ = p;
|
||||
}
|
||||
|
||||
// ---- 通配符转换 ----
|
||||
std::regex HttpServer::wildcardToRegex(const std::string& pattern) {
|
||||
std::string r;
|
||||
r.reserve(pattern.size() * 2);
|
||||
for (char c : pattern) {
|
||||
if (c == '*') r += ".*";
|
||||
else if (c == '?') r += ".";
|
||||
else if (c == '.' || c == '+' || c == '(' || c == ')' || c == '[' || c == ']' ||
|
||||
c == '{' || c == '}' || c == '^' || c == '$' || c == '|' || c == '\\') {
|
||||
r += '\\';
|
||||
r += c;
|
||||
} else r += c;
|
||||
}
|
||||
return std::regex(r, std::regex::icase);
|
||||
}
|
||||
|
||||
bool HttpServer::matchPath(const std::string& path, const std::regex& pattern) {
|
||||
return std::regex_match(path, pattern);
|
||||
}
|
||||
|
||||
// ---- 查询参数解析 ----
|
||||
std::unordered_map<std::string, std::string> HttpServer::parseQueryParams(const std::string& query) {
|
||||
std::unordered_map<std::string, std::string> params;
|
||||
if (query.empty()) return params;
|
||||
|
||||
std::vector<std::string> pairs;
|
||||
boost::split(pairs, query, boost::is_any_of("&"));
|
||||
for (const auto& pair : pairs) {
|
||||
std::vector<std::string> kv;
|
||||
boost::split(kv, pair, boost::is_any_of("="));
|
||||
if (kv.size() == 2) params[kv[0]] = kv[1];
|
||||
else if (kv.size() == 1) params[kv[0]] = "";
|
||||
}
|
||||
return params;
|
||||
}
|
||||
|
||||
// ---- 路由查找 ----
|
||||
HttpRequestHandler HttpServer::findRoute(const std::string& method, const std::string& path) {
|
||||
const std::vector<RouteEntry>* handlers = nullptr;
|
||||
if (method == "GET") handlers = &get_handlers_;
|
||||
else if (method == "POST") handlers = &post_handlers_;
|
||||
else if (method == "PUT") handlers = &put_handlers_;
|
||||
else if (method == "DELETE") handlers = &delete_handlers_;
|
||||
else if (method == "OPTIONS") handlers = &options_handlers_;
|
||||
else return nullptr;
|
||||
|
||||
for (const auto& e : *handlers) {
|
||||
if (!e.is_wildcard && e.path == path) return e.handler;
|
||||
}
|
||||
for (const auto& e : *handlers) {
|
||||
if (e.is_wildcard && matchPath(path, e.pattern)) return e.handler;
|
||||
}
|
||||
return nullptr;
|
||||
}
|
||||
|
||||
bool HttpServer::routeExists(const std::string& method, const std::string& path) {
|
||||
return findRoute(method, path) != nullptr;
|
||||
}
|
||||
|
||||
} // namespace net
|
||||
} // namespace pmc
|
||||
@@ -0,0 +1,106 @@
|
||||
#include "us/detail/HttpServerImpl.hpp"
|
||||
|
||||
namespace pmc {
|
||||
namespace net {
|
||||
|
||||
// ---- WebSocket 路由注册 ----
|
||||
void HttpServer::ws(const std::string& path, WebSocketMessageHandler handler) {
|
||||
WsRouteEntry entry;
|
||||
entry.path = path;
|
||||
entry.handler = 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;
|
||||
}
|
||||
|
||||
void HttpServer::onWsDisconnect(WebSocketDisconnectHandler handler) {
|
||||
ws_disconnect_handler_ = handler;
|
||||
}
|
||||
|
||||
WebSocketMessageHandler HttpServer::findWsRoute(const std::string& path) {
|
||||
for (const auto& e : ws_handlers_) {
|
||||
if (e.path == path) return e.handler;
|
||||
}
|
||||
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_);
|
||||
}
|
||||
|
||||
void HttpServer::addWsClient(const std::string& clientId, std::shared_ptr<Session> session) {
|
||||
std::lock_guard<std::mutex> lock(ws_clients_mutex_);
|
||||
ws_clients_[clientId] = session;
|
||||
}
|
||||
|
||||
void HttpServer::removeWsClient(const std::string& clientId) {
|
||||
std::lock_guard<std::mutex> lock(ws_clients_mutex_);
|
||||
ws_clients_.erase(clientId);
|
||||
}
|
||||
|
||||
std::shared_ptr<HttpServer::Session> HttpServer::getWsClient(const std::string& clientId) {
|
||||
std::lock_guard<std::mutex> lock(ws_clients_mutex_);
|
||||
auto it = ws_clients_.find(clientId);
|
||||
if (it != ws_clients_.end()) return it->second;
|
||||
return nullptr;
|
||||
}
|
||||
|
||||
bool HttpServer::sendToWsClient(const std::string& clientId, const std::string& message) {
|
||||
std::shared_ptr<Session> session;
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(ws_clients_mutex_);
|
||||
auto it = ws_clients_.find(clientId);
|
||||
if (it != ws_clients_.end()) {
|
||||
session = it->second;
|
||||
}
|
||||
}
|
||||
if (session) {
|
||||
session->wsSend(message);
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
size_t HttpServer::broadcastWs(const std::string& message) {
|
||||
std::vector<std::shared_ptr<Session>> clients;
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(ws_clients_mutex_);
|
||||
for (auto& pair : ws_clients_) {
|
||||
clients.push_back(pair.second);
|
||||
}
|
||||
}
|
||||
for (auto& session : clients) {
|
||||
session->wsSend(message);
|
||||
}
|
||||
return clients.size();
|
||||
}
|
||||
|
||||
void HttpServer::disconnectWsClient(const std::string& clientId) {
|
||||
std::lock_guard<std::mutex> lock(ws_clients_mutex_);
|
||||
ws_clients_.erase(clientId);
|
||||
}
|
||||
|
||||
size_t HttpServer::getWsClientCount() const {
|
||||
std::lock_guard<std::mutex> lock(ws_clients_mutex_);
|
||||
return ws_clients_.size();
|
||||
}
|
||||
|
||||
} // namespace net
|
||||
} // namespace pmc
|
||||
@@ -0,0 +1,506 @@
|
||||
#include "us/detail/HttpServerImpl.hpp"
|
||||
|
||||
namespace pmc {
|
||||
namespace net {
|
||||
|
||||
// ===================== Session WebSocket 处理 =====================
|
||||
|
||||
void HttpServer::Session::wsSend(const std::string& message) {
|
||||
auto exec = stream_.get_executor();
|
||||
asio::post(exec, [self = shared_from_this(), message]() {
|
||||
self->ws_send_queue_.push_back(message);
|
||||
if (self->ws_send_queue_.size() == 1) {
|
||||
self->doWsWrite();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
void HttpServer::Session::doWsUpgrade() {
|
||||
auto self = shared_from_this();
|
||||
ws_stream_ = std::make_unique<ws_stream>(stream_);
|
||||
ws_stream_->text(true);
|
||||
|
||||
ws_stream_->async_accept(
|
||||
request_,
|
||||
[self](beast::error_code ec) {
|
||||
self->onWsAccept(ec);
|
||||
});
|
||||
}
|
||||
|
||||
void HttpServer::Session::doWssUpgrade() {
|
||||
auto self = shared_from_this();
|
||||
wss_stream_ = std::make_unique<wss_stream>(*ssl_stream_);
|
||||
wss_stream_->text(true);
|
||||
|
||||
wss_stream_->async_accept(
|
||||
request_,
|
||||
[self](beast::error_code ec) {
|
||||
self->onWsAccept(ec);
|
||||
});
|
||||
}
|
||||
|
||||
void HttpServer::Session::onWsAccept(beast::error_code ec) {
|
||||
if (ec) {
|
||||
std::cerr << "[WS] Accept error from " << remote_ip_ << ":" << remote_port_
|
||||
<< ": " << ec.message() << std::endl;
|
||||
return;
|
||||
}
|
||||
|
||||
is_websocket_ = true;
|
||||
client_id_ = server_->generateWsClientId();
|
||||
server_->addWsClient(client_id_, shared_from_this());
|
||||
|
||||
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());
|
||||
}
|
||||
|
||||
doWsRead();
|
||||
}
|
||||
|
||||
void HttpServer::Session::doWsRead() {
|
||||
auto self = shared_from_this();
|
||||
ws_buffer_.consume(ws_buffer_.size());
|
||||
|
||||
if (ssl_enabled_ && wss_stream_) {
|
||||
wss_stream_->async_read(
|
||||
ws_buffer_,
|
||||
[self](beast::error_code ec, std::size_t bytes_transferred) {
|
||||
self->onWsRead(ec, bytes_transferred);
|
||||
});
|
||||
} else if (ws_stream_) {
|
||||
ws_stream_->async_read(
|
||||
ws_buffer_,
|
||||
[self](beast::error_code ec, std::size_t bytes_transferred) {
|
||||
self->onWsRead(ec, bytes_transferred);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
void HttpServer::Session::onWsRead(beast::error_code ec, std::size_t bytes_transferred) {
|
||||
if (ec == websocket::error::closed) {
|
||||
onWsClose();
|
||||
return;
|
||||
}
|
||||
if (ec) {
|
||||
std::cerr << "[WS] Read error from " << client_id_ << ": " << ec.message() << std::endl;
|
||||
onWsClose();
|
||||
return;
|
||||
}
|
||||
|
||||
std::string message = beast::buffers_to_string(ws_buffer_.data());
|
||||
ws_buffer_.consume(bytes_transferred);
|
||||
|
||||
if (ws_handler_) {
|
||||
auto self = shared_from_this();
|
||||
ws_handler_(client_id_, message,
|
||||
[self](const std::string& response) {
|
||||
self->wsSend(response);
|
||||
});
|
||||
}
|
||||
|
||||
doWsRead();
|
||||
}
|
||||
|
||||
void HttpServer::Session::doWsWrite() {
|
||||
if (ws_send_queue_.empty()) return;
|
||||
|
||||
auto self = shared_from_this();
|
||||
std::string& message = ws_send_queue_.front();
|
||||
|
||||
auto write_handler = [self](beast::error_code ec, std::size_t) {
|
||||
if (ec) {
|
||||
std::cerr << "[WS] Write error to " << self->client_id_ << ": " << ec.message() << std::endl;
|
||||
self->onWsClose();
|
||||
return;
|
||||
}
|
||||
self->ws_send_queue_.pop_front();
|
||||
if (!self->ws_send_queue_.empty()) {
|
||||
self->doWsWrite();
|
||||
}
|
||||
};
|
||||
|
||||
if (ssl_enabled_ && wss_stream_) {
|
||||
wss_stream_->text(true);
|
||||
wss_stream_->async_write(asio::buffer(message), write_handler);
|
||||
} else if (ws_stream_) {
|
||||
ws_stream_->text(true);
|
||||
ws_stream_->async_write(asio::buffer(message), write_handler);
|
||||
}
|
||||
}
|
||||
|
||||
void HttpServer::Session::onWsClose() {
|
||||
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_) {
|
||||
server_->ws_disconnect_handler_(client_id_);
|
||||
}
|
||||
client_id_.clear();
|
||||
is_websocket_ = false;
|
||||
is_ws_proxy_ = false;
|
||||
}
|
||||
|
||||
// ===================== WebSocket Proxy (双向桥接) =====================
|
||||
|
||||
void HttpServer::Session::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 HttpServer::Session::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<tcp::resolver>(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 HttpServer::Session::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<tcp::socket>(exec);
|
||||
proxy_tcp_stream_ = std::make_unique<beast::tcp_stream>(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 HttpServer::Session::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>(asio::ssl::context::tlsv12_client);
|
||||
proxy_ssl_ctx_->set_verify_mode(asio::ssl::verify_none);
|
||||
proxy_ssl_stream_ = std::make_unique<asio::ssl::stream<tcp::socket&>>(
|
||||
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_ws_stream>(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 HttpServer::Session::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_wss_stream>(*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 HttpServer::Session::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 HttpServer::Session::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 HttpServer::Session::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 HttpServer::Session::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 HttpServer::Session::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 HttpServer::Session::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 HttpServer::Session::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 HttpServer::Session::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();
|
||||
}
|
||||
|
||||
} // namespace net
|
||||
} // namespace pmc
|
||||
Reference in new issue
Block a user