From 6692c9c4eed1d9d880e8ca2cf8feaffa726288b5 Mon Sep 17 00:00:00 2001 From: qing <1776177662@qq.com> Date: Tue, 28 Jul 2026 22:21:36 +0800 Subject: [PATCH] =?UTF-8?q?refactor(db):=20=E6=B8=85=E7=90=86=E9=81=97?= =?UTF-8?q?=E7=95=99=E6=AD=BB=E4=BB=A3=E7=A0=81=E5=B9=B6=E5=AF=B9=E9=BD=90?= =?UTF-8?q?=E6=96=87=E6=A1=A3=E4=B8=8E=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 删除 dbv2 从未挂载的 log_module.py（Flask /log/* 遗留）与纯 TCP 服务 用不到的 security_middleware.py 副本；db/README.md 按 pymsg 实现重写， requirements 去掉 flask/waitress；黑盒测试对 pymsg 服务改用 TCP 探测， 修复 check_service HTTP 探测导致的 db 测试整体假 skip。 --- README.md | 1 - db/README.md | 283 +++++++------------- db/log_module.py | 490 ---------------------------------- db/requirements.txt | 2 - db/security_middleware.py | 42 --- tests/blackbox/config.py | 1 + tests/blackbox/helpers.py | 15 ++ tests/blackbox/test_db.py | 106 +------- tests/blackbox/test_runner.py | 8 +- 9 files changed, 130 insertions(+), 818 deletions(-) delete mode 100644 db/log_module.py delete mode 100644 db/security_middleware.py diff --git a/README.md b/README.md index 5d9ebca..c5faffd 100755 --- a/README.md +++ b/README.md @@ -112,7 +112,6 @@ - 外部服务通过 `log_client.py` 以 pymsg 调 execute 路由逐条 INSERT 写入日志。 - `_log_internal()` — db 进程内部直接写 Logs 表，无网络往返，防递归设计。 - 后台守护线程每小时自动清理：保留 30 天 / 上限 100 万条。 - - 注：`db/log_module.py`（Flask `/log/*` Blueprint）为旧版 HTTP 接口的遗留文件，当前 dbv2 未挂载。 - **统一日志客户端 SDK**（各服务目录下的 `log_client.py`，如 [dnfs/log_client.py](dnfs/log_client.py)）：各服务复制即用，纯标准库实现，异步非阻塞（内存队列 + 后台线程，50 条/2 秒批量 flush），fail-safe 设计（db 不可用时静默丢弃，不影响业务）。 ### ai_review（AI 内容审核服务） diff --git a/db/README.md b/db/README.md index 6e1a73c..303817a 100644 --- a/db/README.md +++ b/db/README.md @@ -1,62 +1,55 @@ -# 跨数据库通用查询服务（基于 SQLAlchemy Core） +# 通用数据库服务 db（pymsg/TCP + SQLAlchemy Core） -一个安全的 HTTP 服务，通过结构化 JSON 描述查询意图，后端自动转换为对应数据库的 SQL，无需手写 SQL，即可无缝切换 SQLite、PostgreSQL、MySQL 等数据库。 +一个安全的数据库访问服务。各业务服务通过自研 pymsg 协议（TCP 长度前缀帧 + JSON）发送结构化查询意图，db 翻译为 SQL 执行，全程参数绑定，业务侧无需直连数据库、无需手写 SQL。 ## 功能特点 -- **跨数据库**：基于 SQLAlchemy Core 表达式语言，一套接口适配多种数据库 +- **跨数据库**：基于 SQLAlchemy Core 表达式语言，一套接口适配 SQLite / PostgreSQL / MySQL - **零 SQL 注入**：所有条件、值均通过 SQLAlchemy 参数绑定处理 -- **统一 JSON 接口**：`action` + `table` + `conditions` 模式 -- **支持增删改查**：select, insert, update, delete 全覆盖 -- **单线程本地监听**：waitress + 127.0.0.1:8005，简单可靠 +- **统一 JSON 接口**：`_route` + `action` + `table` + `where` 模式 +- **支持增删改查**：select / insert / update / delete 全覆盖 +- **内置日志表**：启动自动建 Logs 表，供全系统集中写日志；带后台自动清理 +- **仅监听本机**：默认 `127.0.0.1:8005`，不对外暴露 ## 快速开始 ### 安装依赖 + ```bash -pip install flask sqlalchemy waitress -# 根据需要安装数据库驱动，例如： -# pip install psycopg2-binary # PostgreSQL -# pip install pymysql # MySQL +./_set.sh # 建 venv 并安装 requirements.txt（SQLAlchemy + 本地 ../pymsg） ``` -### 启动服务（默认 SQLite） -```bash -python app.py -``` -服务运行在 `http://127.0.0.1:8005`。 +### 启动服务 -使用其他数据库： ```bash -export DATABASE_URL="postgresql://user:pass@localhost/dbname" -python app.py +./_run.sh # venv 中执行 python3 dbv2.py ``` -**注意**：新接口依赖表已存在（通过反射获取结构）。首次使用时请先手动建表： -```bash -sqlite3 database.db "CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, age INTEGER)" -``` +服务监听 `tcp://127.0.0.1:8005`。生产环境由 core（Nidus）按 init.txt 统一拉起，守护靠 systemd。 -## API 文档 +**注意**：`execute` 路由依赖表已存在（通过反射获取结构）。建表见 `create_sqlite3.sql`（Users / Notes；Logs 表由服务启动时自建）。 -### 请求格式 +## 协议与 API -**端点**：`POST /execute` -**Content-Type**：`application/json` +### 传输协议（pymsg） + +每条消息为一个完整 JSON，帧格式：`[4 字节大端长度][UTF-8 JSON]`。 +请求中的 `_route` 字段指定路由，目前仅有一个路由：**`execute`**。 + +### execute 路由 — 通用 CRUD ```json { - "action": "操作类型", // select | insert | update | delete - "table": "表名", - "columns": ["列1", "列2"], // select/insert 时可选 - "values": { // insert/update 时必填 - "列名": "值" - }, - "where": { // 条件，支持组合操作符 + "_route": "execute", + "action": "select", // select | insert | update | delete + "table": "Users", + "columns": ["id", "name"], // select 时可选（缺省全列） + "values": {"name": "Bob"}, // insert / update 时必填 + "where": { // 可选，条件见下表 "age": {"ge": 18, "lt": 60}, "name": "Alice" }, - "order_by": [{"column": "age", "desc": true}], + "order_by": [{"column": "id", "desc": true}], "limit": 10, "offset": 0 } @@ -84,48 +77,46 @@ sqlite3 database.db "CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, age ```json { "success": true, - "data": [ // 查询结果数组（insert/update/delete 时包含影响行数或 lastrowid） - {"id": 1, "name": "Alice"} + "data": [ + {"id": 1, "username": "Alice"} ] } ``` +- select → 行对象数组 +- insert → `[{"lastrowid": N}]` +- update / delete → `[{"rowcount": N}]` 失败： ```json -{ - "success": false, - "error": "错误描述" -} +{"success": false, "error": "错误描述"} ``` -## 使用示例（curl） +## 使用示例（Python + pymsg） -**插入**： -```bash -curl -X POST http://127.0.0.1:8005/execute \ - -H "Content-Type: application/json" \ - -d '{"action":"insert","table":"users","values":{"name":"Bob","age":25}}' -``` +```python +import json +from pymsg import MsgClient -**查询**： -```bash -curl -X POST http://127.0.0.1:8005/execute \ - -H "Content-Type: application/json" \ - -d '{"action":"select","table":"users","where":{"age":{"gt":20}},"order_by":[{"column":"id","desc":true}],"limit":5}' -``` +client = MsgClient("127.0.0.1", 8005, "myservice") -**更新**： -```bash -curl -X POST http://127.0.0.1:8005/execute \ - -H "Content-Type: application/json" \ - -d '{"action":"update","table":"users","values":{"age":26},"where":{"name":"Bob"}}' -``` +# 查询 +resp = client.call(json.dumps({ + "_route": "execute", + "action": "select", + "table": "Users", + "where": {"id": {"gt": 0}}, + "order_by": [{"column": "id", "desc": True}], + "limit": 5, +})) +print(json.loads(resp)) -**删除**： -```bash -curl -X POST http://127.0.0.1:8005/execute \ - -H "Content-Type: application/json" \ - -d '{"action":"delete","table":"users","where":{"id":{"in":[1,2]}}}' +# 插入 +client.call(json.dumps({ + "_route": "execute", + "action": "insert", + "table": "Users", + "values": {"username": "bob", "password_hash": "..."}, +})) ``` ## 安全机制 @@ -133,17 +124,17 @@ curl -X POST http://127.0.0.1:8005/execute \ - 强制使用 SQLAlchemy 参数绑定，杜绝 SQL 注入。 - 仅反射已有表，无法创建或删除表（保证结构安全）。 - 所有异常自动回滚事务。 -- 默认监听 `127.0.0.1`，仅本机可访问。 +- 仅监听 `127.0.0.1`，仅本机服务可访问。 -## 内置日志模块 +## 内置日志能力 -db 服务内置了统一日志收集能力，各业务服务通过 HTTP 推送日志到 `/log/*` 系列接口，db 自身的操作日志通过 `log_internal()` 直接写入（同进程，无 HTTP 往返，无循环依赖）。 +db 内置统一日志收集：全系统各服务（waf / gateway / static / dnfs / db / ai_review / captcha / core / smtp）的日志集中写入 db 的 `Logs` 表。 ### 为什么集成在 db 而不是独立服务 独立 logs 服务会导致**循环调用**：db 记日志→调 logs，logs 存库→调 db，无限递归。集成在 db 内部后，db 自己记日志直接写表，彻底消除循环依赖。 -### 日志表 +### 日志表（启动时自建） ```sql CREATE TABLE IF NOT EXISTS Logs ( @@ -155,152 +146,66 @@ CREATE TABLE IF NOT EXISTS Logs ( trace_id TEXT, extra TEXT ); +-- 索引: (service,ts) / (level,ts) / (ts) / (trace_id) ``` -- `service`：来源服务名（`waf`/`gateway`/`static`/`dnfs`/`db`/`ai_review`/`captcha`/`core`） -- `level`：`DEBUG`/`INFO`/`WARNING`/`ERROR`/`CRITICAL` +- `level`：`DEBUG` / `INFO` / `WARNING` / `ERROR` / `CRITICAL` - `trace_id`：可选，请求追踪 ID - `extra`：可选，JSON 字符串，存放附加结构化字段 -### 接口 +### 写入方式 -| 接口 | 方法 | 说明 | -|------|------|------| -| `/log/ingest` | POST | 推送日志（支持批量） | -| `/log/query` | POST | 查询日志（按服务/级别/时间/关键字过滤） | -| `/log/stats` | POST | 统计聚合（按 service/level 分组计数） | -| `/log/purge` | POST | 清理过期日志 | -| `/log/health` | GET | 健康检查 | +**外部服务**：复制 `example_log_client.py` 为各自目录下的 `log_client.py`，改 `service_name` 即用。客户端底层通过 pymsg 调 `execute` 路由逐条 INSERT 到 Logs 表： -所有接口的响应格式与 `/execute` 统一：`{success, data, error}`。 +```python +from log_client import logs -#### POST /log/ingest - -```json -请求: -{ - "logs": [ - { - "service": "dnfs", - "level": "INFO", - "message": "节点新增成功 id=42", - "ts": "2026-07-25T08:30:00.123Z", - "trace_id": "req-abc-123", - "extra": {"path": "/add_note", "status": 200} - } - ] -} - -响应: -{"success": true, "data": {"inserted": 1}} +logs.info("dnfs", "节点新增成功 id=42") +logs.error("dnfs", "数据库连接失败", extra={"errno": 2002}) +logs.push("gateway", "WARNING", "登录失败次数超限", + trace_id="req-abc", extra={"ip": "1.2.3.4"}) ``` -- `ts` 可选，缺省自动填当前 UTC 时间 -- 单次最多 100 条 -- `message` 上限 8KB，`extra` 上限 4KB +客户端特性： +- 纯标准库实现，异步非阻塞：内存队列 + 后台线程批量推送 +- fail-safe：db 不可用时静默丢弃，绝不影响业务 +- 缓冲批量：队列满 50 条或超时 2 秒时自动 flush -#### POST /log/query +**db 自身**：`_log_internal()` 直接写库，无网络往返（防递归）。 -```json -请求: -{ - "service": "dnfs", // 可选，支持数组匹配多个 - "level": "WARNING", // 可选，支持数组匹配多个 - "since": "2026-07-25T00:00:00Z", // 可选 - "until": "2026-07-25T23:59:59Z", // 可选 - "keyword": "节点", // 可选，message 模糊搜索 - "trace_id": "req-abc-123", // 可选 - "order_by": "ts", // 默认 ts，可选 id/service/level - "desc": true, // 默认 true（最新在前） - "limit": 50, // 默认 100，上限 1000 - "offset": 0 // 分页偏移 -} +### 查询日志 -响应: -{ - "success": true, - "data": { - "total": 128, - "logs": [{"id":1024,"ts":"...","service":"dnfs","level":"INFO","message":"...","trace_id":"...","extra":{...}}] - } -} -``` +日志查询直接走 `execute` 路由 select Logs 表，或用工具脚本 `_view_db.sqlite3.py`： -#### POST /log/stats - -```json -请求: -{"since":"...","until":"...","group_by":"service"} - -响应: -{ - "success": true, - "data": { - "groups": [{"service":"dnfs","level":"INFO","count":320}], - "total": 325 - } -} -``` - -`group_by` 支持 `service` / `level` / `service,level`。 - -#### POST /log/purge - -```json -请求: {"before": "2026-07-18T00:00:00Z"} -响应: {"success": true, "data": {"deleted": 1523}} -``` - -#### GET /log/health - -```json -{"success": true, "data": {"status": "ok", "count": 10240}} +```bash +python3 _view_db.sqlite3.py # 列出所有表及行数 +python3 _view_db.sqlite3.py Logs -n 20 # 最近 20 条日志 +python3 _view_db.sqlite3.py Logs -w "level='ERROR'" +python3 _view_db.sqlite3.py -s "SELECT service, COUNT(*) FROM Logs GROUP BY service" ``` ### 自动清理 - 服务启动时开启后台守护线程，每 1 小时清理一次 - 默认保留 30 天日志，超期自动删除 -- 日志总条数超过 100 万时，按时间倒序保留最新 N 条 - -### 各服务接入方式 - -把 `db/log_client.py` 复制到各服务目录（或加入 `PYTHONPATH`），即可使用： - -```python -from log_client import logs - -# 替换原来的 print -logs.info("dnfs", "节点新增成功 id=42") -logs.error("dnfs", "数据库连接失败", extra={"errno": 2002}) - -# 带追踪 ID -logs.push("gateway", "WARNING", "登录失败次数超限", - trace_id="req-abc", extra={"ip": "1.2.3.4"}) -``` - -特性： -- 使用 `urllib.request`（标准库，不引入 requests） -- 异步非阻塞：内存队列 + 后台线程批量推送 -- fail-safe：db 服务不可用时静默丢弃，绝不影响业务 -- 缓冲批量：队列满 50 条或超时 2 秒时自动 flush - -### 防递归设计 - -| 场景 | 调用路径 | 是否安全 | -|------|---------|---------| -| 外部服务记日志 | `服务 → POST /log/ingest → 写 Logs 表` | 安全 | -| db 自己记日志 | `/execute → log_internal() → 直接写 Logs 表` | 安全（无 HTTP） | -| /log/ingest 成功 | `写入成功 → 直接返回` | 安全（不再记日志） | -| /log/ingest 失败 | `写入失败 → log_internal() → 直接写表` | 安全（log_internal 不回调 ingest） | +- 日志总条数超过 100 万时，按时间倒序保留最新 100 万条 ## 切换数据库 只需修改环境变量 `DATABASE_URL`，并安装对应驱动： -- SQLite：`sqlite:///database.db` -- PostgreSQL：`postgresql://user:pass@localhost/dbname` （需 `pip install psycopg2`） -- MySQL：`mysql+pymysql://user:pass@localhost/dbname` （需 `pip install pymysql`） +- SQLite（默认）：`sqlite:///database.db` +- PostgreSQL：`postgresql://user:pass@localhost/dbname`（需 `pip install psycopg2`） +- MySQL：`mysql+pymysql://user:pass@localhost/dbname`（需 `pip install pymysql`） SQLAlchemy 会自动生成适配各数据库的 SQL 方言，无需修改接口调用。 +## 文件说明 + +| 文件 | 说明 | +|------|------| +| `dbv2.py` | 服务主程序（pymsg/TCP，唯一在跑的实现） | +| `example_log_client.py` | 日志客户端 SDK 模板，各服务复制为 `log_client.py` 使用 | +| `create_sqlite3.sql` | Users / Notes / Logs 建表脚本 | +| `_view_db.sqlite3.py` | 非交互式查看 database.db 的工具脚本 | +| `_set.sh` / `_run.sh` | 环境安装 / 启动脚本 | diff --git a/db/log_module.py b/db/log_module.py deleted file mode 100644 index 385c387..0000000 --- a/db/log_module.py +++ /dev/null @@ -1,490 +0,0 @@ -""" -db 服务的内置日志子模块。 - -提供两部分能力： - 1. log_internal() — db 进程内部直接写 Logs 表（无 HTTP 往返，无循环依赖） - 2. /log/* HTTP 路由 — 供外部服务推送 / 查询 / 统计 / 清理日志 - -设计要点（防递归）： - - log_internal() 直接操作数据库，不经过任何 HTTP 路由 - - /log/ingest 成功后不再调 log_internal - - /log/ingest 失败时才调 log_internal，而 log_internal 不会回调 /log/ingest -""" - -import json -import threading -import time -from datetime import datetime, timezone, timedelta - -from flask import Blueprint, request, jsonify, current_app -from sqlalchemy import ( - Table, insert, select, delete, func, and_, or_, MetaData -) - -bp = Blueprint('log', __name__, url_prefix='/log') - -# ── 常量 ────────────────────────────────────────────────────────────── - -VALID_SERVICES = { - 'waf', 'gateway', 'static', 'dnfs', 'db', - 'ai_review', 'captcha', 'core', -} -VALID_LEVELS = {'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'} - -MAX_BATCH = 100 # 单次 /log/ingest 最多 100 条 -MAX_MESSAGE_LEN = 8192 # message 上限 8KB -MAX_EXTRA_LEN = 4096 # extra 上限 4KB - -DEFAULT_RETENTION_DAYS = 30 -DEFAULT_MAX_RECORDS = 1_000_000 -DEFAULT_CLEANUP_INTERVAL = 3600 # 秒 - -# ── 表管理 ───────────────────────────────────────────────────────────── - -_LOGS_TABLE_CACHE = {} - -def _ensure_logs_table(metadata: MetaData, engine): - """确保 Logs 表已建好并已反射到 metadata（首次调用时建表 + 反射）。""" - if 'Logs' in _LOGS_TABLE_CACHE: - return _LOGS_TABLE_CACHE['Logs'] - - # 先用原生 SQL 建表（SQLAlchemy 反射要求表已存在） - with engine.connect() as conn: - conn.exec_driver_sql(""" - CREATE TABLE IF NOT EXISTS Logs ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - ts TEXT NOT NULL, - service TEXT NOT NULL, - level TEXT NOT NULL, - message TEXT NOT NULL, - trace_id TEXT, - extra TEXT - ) - """) - conn.exec_driver_sql( - "CREATE INDEX IF NOT EXISTS idx_logs_service_ts ON Logs(service, ts)" - ) - conn.exec_driver_sql( - "CREATE INDEX IF NOT EXISTS idx_logs_level_ts ON Logs(level, ts)" - ) - conn.exec_driver_sql( - "CREATE INDEX IF NOT EXISTS idx_logs_ts ON Logs(ts)" - ) - conn.exec_driver_sql( - "CREATE INDEX IF NOT EXISTS idx_logs_trace ON Logs(trace_id)" - ) - conn.commit() - - # 反射到 metadata - table = Table('Logs', metadata, autoload_with=engine) - _LOGS_TABLE_CACHE['Logs'] = table - return table - - -# ── 内部日志记录（db 自己用，无 HTTP）───────────────────────────────── - -def log_internal(engine, metadata, service, level, message, - trace_id=None, extra=None): - """ - db 进程内部记录日志 — 直接操作数据库，不走 HTTP，无循环依赖。 - - 供 db 的 /execute 路由内部调用。失败时静默吞掉，不影响主业务。 - """ - try: - table = _ensure_logs_table(metadata, engine) - extra_json = json.dumps(extra, ensure_ascii=False) if extra else None - if len(message) > MAX_MESSAGE_LEN: - message = message[:MAX_MESSAGE_LEN] - if extra_json and len(extra_json) > MAX_EXTRA_LEN: - extra_json = extra_json[:MAX_EXTRA_LEN] - - with engine.connect() as conn: - conn.execute(insert(table).values( - ts=datetime.now(timezone.utc).isoformat(), - service=service, - level=level, - message=message, - trace_id=trace_id, - extra=extra_json, - )) - conn.commit() - except Exception: - # 日志记录失败不应影响主业务 - pass - - -# ── HTTP 路由 ────────────────────────────────────────────────────────── - -def init_log_routes(app, engine, metadata, - retention_days=DEFAULT_RETENTION_DAYS, - max_records=DEFAULT_MAX_RECORDS, - cleanup_interval=DEFAULT_CLEANUP_INTERVAL): - """把日志蓝图注册到 Flask app，并注入 engine/metadata。""" - - app.config['LOG_ENGINE'] = engine - app.config['LOG_METADATA'] = metadata - app.config['LOG_RETENTION_DAYS'] = retention_days - app.config['LOG_MAX_RECORDS'] = max_records - app.config['LOG_CLEANUP_INTERVAL'] = cleanup_interval - - # 建表 + 反射 - try: - _ensure_logs_table(metadata, engine) - except Exception as e: - log_internal(engine, metadata, 'db', 'ERROR', - f'Logs 表初始化失败: {e}') - - # 注册路由 - app.register_blueprint(bp) - - # 启动后台清理线程 - _start_cleanup_thread(engine, metadata, - retention_days, max_records, cleanup_interval) - - -# ── 5.1 推送日志 ────────────────────────────────────────────────────── - -@bp.route('/ingest', methods=['POST']) -def ingest(): - """接收外部日志推送（单条或批量）。""" - data = request.get_json(silent=True) - if not data or 'logs' not in data: - return jsonify({'success': False, 'error': '请求体缺少 logs 字段'}), 400 - - logs = data['logs'] - if not isinstance(logs, list) or not logs: - return jsonify({'success': False, 'error': 'logs 必须是非空数组'}), 400 - if len(logs) > MAX_BATCH: - return jsonify({ - 'success': False, - 'error': f'单次最多 {MAX_BATCH} 条，当前 {len(logs)} 条' - }), 400 - - engine = current_app.config['LOG_ENGINE'] - metadata = current_app.config['LOG_METADATA'] - - try: - table = _ensure_logs_table(metadata, engine) - except Exception as e: - return jsonify({'success': False, 'error': f'表初始化失败: {e}'}), 500 - - # 校验 + 组装 - now = datetime.now(timezone.utc).isoformat() - rows = [] - for i, log in enumerate(logs): - if not isinstance(log, dict): - return jsonify({'success': False, - 'error': f'第 {i} 条日志不是有效对象'}), 400 - - service = log.get('service') - level = log.get('level') - message = log.get('message') - - if not service or service not in VALID_SERVICES: - return jsonify({'success': False, - 'error': f'第 {i} 条: service 无效或缺失'}), 400 - if not level or level not in VALID_LEVELS: - return jsonify({'success': False, - 'error': f'第 {i} 条: level 无效或缺失'}), 400 - if not message or not isinstance(message, str): - return jsonify({'success': False, - 'error': f'第 {i} 条: message 缺失或非字符串'}), 400 - - if len(message) > MAX_MESSAGE_LEN: - message = message[:MAX_MESSAGE_LEN] - - trace_id = log.get('trace_id') - extra = log.get('extra') - extra_json = None - if extra is not None: - extra_json = json.dumps(extra, ensure_ascii=False) - if len(extra_json) > MAX_EXTRA_LEN: - extra_json = extra_json[:MAX_EXTRA_LEN] - - rows.append({ - 'ts': log.get('ts') or now, - 'service': service, - 'level': level, - 'message': message, - 'trace_id': trace_id, - 'extra': extra_json, - }) - - # 写入（成功后不再调 log_internal，防递归） - try: - with engine.connect() as conn: - conn.execute(insert(table), rows) - conn.commit() - except Exception as e: - # 失败时可以安全地记日志（log_internal 不回调 ingest） - log_internal(engine, metadata, 'db', 'ERROR', - f'/log/ingest 写入失败: {e}') - return jsonify({'success': False, 'error': str(e)}), 500 - - return jsonify({'success': True, 'data': {'inserted': len(rows)}}) - - -# ── 5.2 查询日志 ─────────────────────────────────────────────────────── - -@bp.route('/query', methods=['POST']) -def query(): - """结构化查询日志。""" - data = request.get_json(silent=True) or {} - - engine = current_app.config['LOG_ENGINE'] - metadata = current_app.config['LOG_METADATA'] - - try: - table = _ensure_logs_table(metadata, engine) - except Exception as e: - return jsonify({'success': False, 'error': f'表初始化失败: {e}'}), 500 - - clauses = [] - - # service（支持单个字符串或数组） - svc = data.get('service') - if svc: - col = table.columns['service'] - if isinstance(svc, list): - clauses.append(col.in_(svc)) - else: - clauses.append(col == svc) - - # level（支持单个字符串或数组） - lvl = data.get('level') - if lvl: - col = table.columns['level'] - if isinstance(lvl, list): - clauses.append(col.in_(lvl)) - else: - clauses.append(col == lvl) - - # trace_id - tid = data.get('trace_id') - if tid: - clauses.append(table.columns['trace_id'] == tid) - - # 时间区间 - since = data.get('since') - until = data.get('until') - ts_col = table.columns['ts'] - if since: - clauses.append(ts_col >= since) - if until: - clauses.append(ts_col <= until) - - # 关键字模糊搜索 - keyword = data.get('keyword') - if keyword: - clauses.append(table.columns['message'].like(f'%{keyword}%')) - - where_cond = and_(*clauses) if clauses else None - - # 排序 - order_col_name = data.get('order_by', 'ts') - if order_col_name not in ('id', 'ts', 'service', 'level'): - order_col_name = 'ts' - order_col = table.columns[order_col_name] - desc = data.get('desc', True) - order_expr = order_col.desc() if desc else order_col.asc() - - # 分页 - limit = min(data.get('limit', 100), 1000) - offset = data.get('offset', 0) - - try: - with engine.connect() as conn: - # 总数 - count_stmt = select(func.count()).select_from(table) - if where_cond is not None: - count_stmt = count_stmt.where(where_cond) - total = conn.execute(count_stmt).scalar() or 0 - - # 查询 - stmt = select(table) - if where_cond is not None: - stmt = stmt.where(where_cond) - stmt = stmt.order_by(order_expr).limit(limit).offset(offset) - - rows = conn.execute(stmt).fetchall() - logs = [] - for row in rows: - d = dict(row._mapping) - if d.get('extra'): - try: - d['extra'] = json.loads(d['extra']) - except (json.JSONDecodeError, TypeError): - pass - logs.append(d) - - return jsonify({'success': True, 'data': {'total': total, 'logs': logs}}) - - except Exception as e: - log_internal(engine, metadata, 'db', 'ERROR', - f'/log/query 失败: {e}') - return jsonify({'success': False, 'error': str(e)}), 500 - - -# ── 5.3 统计聚合 ─────────────────────────────────────────────────────── - -@bp.route('/stats', methods=['POST']) -def stats(): - """日志分类统计。""" - data = request.get_json(silent=True) or {} - - engine = current_app.config['LOG_ENGINE'] - metadata = current_app.config['LOG_METADATA'] - - try: - table = _ensure_logs_table(metadata, engine) - except Exception as e: - return jsonify({'success': False, 'error': f'表初始化失败: {e}'}), 500 - - group_by = data.get('group_by', 'service') - group_fields = [g.strip() for g in group_by.split(',') if g.strip()] - for g in group_fields: - if g not in ('service', 'level'): - return jsonify({ - 'success': False, - 'error': f'group_by 仅支持 service / level，收到: {g}' - }), 400 - - # 时间过滤 - clauses = [] - since = data.get('since') - until = data.get('until') - ts_col = table.columns['ts'] - if since: - clauses.append(ts_col >= since) - if until: - clauses.append(ts_col <= until) - where_cond = and_(*clauses) if clauses else None - - try: - with engine.connect() as conn: - cols = [table.columns[g] for g in group_fields] + [func.count().label('count')] - stmt = select(*cols) - if where_cond is not None: - stmt = stmt.where(where_cond) - stmt = stmt.group_by(*[table.columns[g] for g in group_fields]) - - rows = conn.execute(stmt).fetchall() - groups = [dict(row._mapping) for row in rows] - - # 总数 - count_stmt = select(func.count()).select_from(table) - if where_cond is not None: - count_stmt = count_stmt.where(where_cond) - total = conn.execute(count_stmt).scalar() or 0 - - return jsonify({'success': True, 'data': {'groups': groups, 'total': total}}) - - except Exception as e: - log_internal(engine, metadata, 'db', 'ERROR', - f'/log/stats 失败: {e}') - return jsonify({'success': False, 'error': str(e)}), 500 - - -# ── 5.4 清理过期 ────────────────────────────────────────────────────── - -@bp.route('/purge', methods=['POST']) -def purge(): - """清理过期日志。""" - data = request.get_json(silent=True) or {} - before = data.get('before') - if not before: - return jsonify({'success': False, 'error': '缺少 before 字段'}), 400 - - engine = current_app.config['LOG_ENGINE'] - metadata = current_app.config['LOG_METADATA'] - - try: - table = _ensure_logs_table(metadata, engine) - except Exception as e: - return jsonify({'success': False, 'error': f'表初始化失败: {e}'}), 500 - - try: - with engine.connect() as conn: - stmt = delete(table).where(table.columns['ts'] < before) - result = conn.execute(stmt) - conn.commit() - deleted = result.rowcount - log_internal(engine, metadata, 'db', 'INFO', - f'/log/purge 清理 {deleted} 条 (before={before})') - return jsonify({'success': True, 'data': {'deleted': deleted}}) - - except Exception as e: - log_internal(engine, metadata, 'db', 'ERROR', - f'/log/purge 失败: {e}') - return jsonify({'success': False, 'error': str(e)}), 500 - - -# ── 5.5 健康检查 ────────────────────────────────────────────────────── - -@bp.route('/health', methods=['GET']) -def health(): - """日志模块健康检查。""" - engine = current_app.config['LOG_ENGINE'] - metadata = current_app.config['LOG_METADATA'] - - try: - table = _ensure_logs_table(metadata, engine) - with engine.connect() as conn: - count = conn.execute(select(func.count()).select_from(table)).scalar() or 0 - return jsonify({'success': True, 'data': {'status': 'ok', 'count': count}}) - except Exception as e: - return jsonify({'success': False, 'error': str(e)}), 500 - - -# ── 后台自动清理线程 ────────────────────────────────────────────────── - -def _start_cleanup_thread(engine, metadata, - retention_days=DEFAULT_RETENTION_DAYS, - max_records=DEFAULT_MAX_RECORDS, - interval=DEFAULT_CLEANUP_INTERVAL): - """启动后台守护线程，定期清理过期日志。""" - def _cleanup_loop(): - while True: - time.sleep(interval) - try: - _do_cleanup(engine, metadata, retention_days, max_records) - except Exception: - # 清理失败不影响服务运行 - pass - - t = threading.Thread(target=_cleanup_loop, daemon=True, name='logs-cleanup') - t.start() - - -def _do_cleanup(engine, metadata, retention_days, max_records): - """执行一次清理：删过期 + 裁剪超量。""" - table = _ensure_logs_table(metadata, engine) - - # 1. 删除过期日志 - cutoff = (datetime.now(timezone.utc) - timedelta(days=retention_days)).isoformat() - with engine.connect() as conn: - result = conn.execute(delete(table).where(table.columns['ts'] < cutoff)) - conn.commit() - expired_deleted = result.rowcount - - # 2. 裁剪超量记录（保留最新的 max_records 条） - count = conn.execute(select(func.count()).select_from(table)).scalar() or 0 - trimmed = 0 - if count > max_records: - excess = count - max_records - # 找到第 max_records 条的 ts（按时间正序的第 max_records 条） - cutoff_row = conn.execute( - select(table.columns['ts']) - .order_by(table.columns['ts'].asc()) - .limit(1).offset(max_records) - ).fetchone() - if cutoff_row: - trim_result = conn.execute( - delete(table).where(table.columns['ts'] < cutoff_row[0]) - ) - conn.commit() - trimmed = trim_result.rowcount - - if expired_deleted or trimmed: - log_internal(engine, metadata, 'db', 'INFO', - f'自动清理: 过期删除 {expired_deleted} 条, 超量裁剪 {trimmed} 条') diff --git a/db/requirements.txt b/db/requirements.txt index be8d7ed..390bf2e 100644 --- a/db/requirements.txt +++ b/db/requirements.txt @@ -1,5 +1,3 @@ nuitka -flask SQLAlchemy -waitress ../pymsg \ No newline at end of file diff --git a/db/security_middleware.py b/db/security_middleware.py deleted file mode 100644 index 8f69800..0000000 --- a/db/security_middleware.py +++ /dev/null @@ -1,42 +0,0 @@ -""" -Security headers middleware for Flask applications. -Usage: - from security_middleware import apply_security_headers - apply_security_headers(app) -""" -import secrets - -def apply_security_headers(app, csp_policy=None): - """Apply security response headers to a Flask app.""" - - @app.after_request - def set_security_headers(response): - # Prevent clickjacking - response.headers['X-Frame-Options'] = 'SAMEORIGIN' - # XSS protection (legacy browsers) - response.headers['X-XSS-Protection'] = '1; mode=block' - # Prevent MIME type sniffing - response.headers['X-Content-Type-Options'] = 'nosniff' - # Referrer policy - response.headers['Referrer-Policy'] = 'strict-origin-when-cross-origin' - # Permissions policy (disable camera, mic, geolocation) - response.headers['Permissions-Policy'] = 'camera=(), microphone=(), geolocation=()' - - # Content Security Policy - if csp_policy: - response.headers['Content-Security-Policy'] = csp_policy - else: - response.headers['Content-Security-Policy'] = ( - "default-src 'self'; " - "script-src 'self' 'unsafe-inline'; " - "style-src 'self' 'unsafe-inline'; " - "img-src 'self' data: blob:; " - "font-src 'self'; " - "connect-src 'self'; " - "frame-ancestors 'self'" - ) - - # Remove server header info - response.headers.pop('Server', None) - - return response diff --git a/tests/blackbox/config.py b/tests/blackbox/config.py index 1854538..22e5450 100644 --- a/tests/blackbox/config.py +++ b/tests/blackbox/config.py @@ -31,6 +31,7 @@ SERVICES = { "name": "db", "host": "127.0.0.1", "port": 8005, + "protocol": "pymsg", "description": "通用数据库服务 (CRUD+集中日志)", "test_file": "test_db.py", }, diff --git a/tests/blackbox/helpers.py b/tests/blackbox/helpers.py index 6509873..b4ba71e 100644 --- a/tests/blackbox/helpers.py +++ b/tests/blackbox/helpers.py @@ -284,6 +284,21 @@ class PymsgClient: # 服务可用性检查 # --------------------------------------------------------------------------- +def check_tcp_service(host: str, port: int, timeout: float = 5.0) -> tuple: + """检查纯 TCP 服务（如 pymsg 协议服务 db/captcha/smtp）是否可达。 + 返回 (reachable: bool, detail: str)。 + """ + try: + with socket.create_connection((host, port), timeout=timeout): + return True, "TCP 连接成功" + except ConnectionRefusedError: + return False, "连接被拒绝 (服务未启动或端口未监听)" + except socket.timeout: + return False, f"TCP 连接超时 ({timeout}s)" + except OSError as e: + return False, f"网络错误: {e}" + + def check_service(url: str, timeout: float = 8.0) -> tuple: """检查服务是否可达（HTTP 请求检测，与测试实际行为一致）。 返回 (reachable: bool, detail: str)。 diff --git a/tests/blackbox/test_db.py b/tests/blackbox/test_db.py index ca922b2..4be4a96 100644 --- a/tests/blackbox/test_db.py +++ b/tests/blackbox/test_db.py @@ -1,45 +1,37 @@ """ -db 黑盒用户模拟测试 — 集中日志: 写入 → 查询 → 统计 → 清理。 +db 黑盒用户模拟测试 — pymsg 协议 CRUD。 用户旅程: - 1. 通过 pymsg 协议执行 CRUD (select/insert) - 2. 通过 HTTP /log/ingest 写入日志 - 3. 通过 HTTP /log/query 查询日志 - 4. 通过 HTTP /log/stats 查看统计 - 5. 通过 HTTP /log/health 健康检查 + 1. 通过 pymsg 协议查询 Users 表 + 2. 通过 pymsg 协议查询 Logs 表 + 3. 通过 pymsg 协议带 where 条件查询 + 4. 无效路由被拒绝 -注意: dbv2.py (pymsg 模式) 不挂载 Flask log_module Blueprint， -因此 HTTP /log/* 路由可能不可用。测试会优雅降级。 +注意: dbv2.py 为纯 pymsg/TCP 服务（无 HTTP 接口），日志通过 +log_client.py 走 execute 路由 INSERT 写入 Logs 表。 """ -import json -import random -import string -import time - from helpers import ( - http_get, http_post, try_parse_json, - assert_eq, assert_status, assert_true, assert_json_key, - skip, Counter, check_service, PymsgClient, + assert_true, skip, Counter, PymsgClient, check_tcp_service, ) -from config import get_url -DB_URL = get_url("db") DB_HOST = "127.0.0.1" DB_PORT = 8005 def run(url: str = None): - """db 用户模拟测试入口。返回 (passed, failed, skipped)。""" - du = url or DB_URL + """db 用户模拟测试入口。返回 (passed, failed, skipped)。 + + url 参数仅为保持与其他测试模块的签名一致，db 固定走 127.0.0.1:8005。 + """ c = Counter() print(f"\n--- db 数据库与日志流程 ---\n") # ------------------------------------------------------------------ - # 0. 检查服务可用性 + # 0. 检查服务可用性（TCP 直连，db 无 HTTP 接口） # ------------------------------------------------------------------ - if not check_service(du)[0]: + if not check_tcp_service(DB_HOST, DB_PORT)[0]: skip("全部测试", "db 服务不可达", c) return c.passed, c.failed, c.skipped @@ -109,76 +101,6 @@ def run(url: str = None): "pymsg-无效路由-拒绝", c, ) - # ------------------------------------------------------------------ - # 第二部分: HTTP /log/* 接口 (Flask log_module) - # ------------------------------------------------------------------ - print("\n [HTTP /log/* 接口测试]") - - # 检查 /log/health 是否可用（Flask log_module 是否挂载） - status, body, _ = http_get(f"{du}/log/health") - if status == 0 or status >= 500: - skip("HTTP /log/* 测试", "log_module 未挂载 (dbv2 使用 pymsg 模式, 非 Flask)", c) - else: - # 场景 5: /log/health 健康检查 - assert_eq(status, 200, "log-健康检查 (200)", c) - - # 场景 6: /log/ingest 写入日志 - test_ts = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) - status, body, _ = http_post(f"{du}/log/ingest", { - "logs": [{ - "service": "dnfs", - "level": "INFO", - "message": f"blackbox_test_{random.randint(10000, 99999)}", - "ts": test_ts, - }], - }) - data = try_parse_json(body) - assert_true( - status == 200 and data is not None and data.get("success") is True, - "log-写入日志", c, - f"status={status} body={body[:150]}", - ) - - # 场景 7: /log/ingest — 空日志列表 - status, body, _ = http_post(f"{du}/log/ingest", {"logs": []}) - assert_eq(status, 200, "log-空日志列表 (200)", c) - - # 场景 8: /log/ingest — 缺少 logs 字段 - status, body, _ = http_post(f"{du}/log/ingest", {}) - data = try_parse_json(body) - assert_true( - data is not None and not data.get("success", True), - "log-缺少logs字段-拒绝", c, - ) - - # 场景 9: /log/query 查询日志 - status, body, _ = http_post(f"{du}/log/query", { - "limit": 5, - "desc": True, - }) - data = try_parse_json(body) - assert_true( - status == 200 and data is not None and data.get("success") is True, - "log-查询日志", c, - f"status={status}", - ) - if data and data.get("success"): - assert_true( - "logs" in data.get("data", {}), - "log-查询返回logs字段", c, - ) - - # 场景 10: /log/stats 统计 - status, body, _ = http_post(f"{du}/log/stats", { - "group_by": "service", - }) - data = try_parse_json(body) - assert_true( - status == 200 and data is not None, - "log-统计", c, - f"status={status}", - ) - # ------------------------------------------------------------------ # 汇总 # ------------------------------------------------------------------ diff --git a/tests/blackbox/test_runner.py b/tests/blackbox/test_runner.py index 8729763..8f1ff84 100644 --- a/tests/blackbox/test_runner.py +++ b/tests/blackbox/test_runner.py @@ -17,7 +17,7 @@ import sys sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from config import SERVICES, PORT_MAP, detect_service, get_url -from helpers import check_service +from helpers import check_service, check_tcp_service TEST_MODULES = { "gateway": "test_gateway", @@ -53,7 +53,11 @@ def run_service_test(name: str, url: str) -> tuple: print(f" {svc['description']} ({url})") print(f"{'='*60}") - reachable, detail = check_service(url) + # pymsg/TCP 服务无 HTTP 接口，用 TCP 直连探测 + if svc.get("protocol") == "pymsg": + reachable, detail = check_tcp_service(svc["host"], svc["port"]) + else: + reachable, detail = check_service(url) if not reachable: print(f" [WARN] 服务不可达: {url}") if detail: