# md-vector-db 架构审计报告

**审计日期**: 2026-07-05
**审计范围**: 全量代码 (src/, tests/, config.yaml, pyproject.toml)
**审计员**: 架构评审工程师
**严重等级定义**:

| 等级 | 符号 | 含义 |
|------|------|------|
| **严重** | 🔴 | 可导致数据泄露、服务崩溃或安全漏洞，须立即修复 |
| **高** | 🟠 | 可能导致线上故障、数据不一致或性能严重退化 |
| **中** | 🟡 | 影响可维护性、可测试性或存在潜在风险 |
| **低** | 🔵 | 代码风格、命名或次要改进建议 |

---

## 一、安全性 (Security)

### 🔴 S-01: API 密钥明文存储

**文件**: `config.yaml:9`, `src/core/config.py:25`

```yaml
embed:
  api_key: ""  # api 模式下填写
```

API 密钥直接存在 YAML 文件中，任何能读取项目目录的人都能获取。`config.yaml` 虽然列在 `.gitignore` 的 `data/` 规则之外，但目前文件已在 git 追踪中。

**修复建议**: 从 `config.yaml` 移除 `api_key` 字段，改为从环境变量 `EMBED_API_KEY` 读取：

```python
@dataclass
class EmbedConfig:
    api_key: str = field(default_factory=lambda: os.getenv("EMBED_API_KEY", ""))
```

---

### 🔴 S-02: 路径遍历漏洞

**文件**: `src/server/app.py:111-114`

```python
if req.file_path:
    path = Path(req.file_path)
    if not path.exists():
        raise HTTPException(status_code=404, ...)
    count = ingestor.ingest_file(str(path))
```

`file_path` 参数无任何校验，攻击者可通过 `/api/v1/ingest` 传入 `../../../etc/passwd` 读取服务器任意文件。虽然 Windows 上影响有限，但若部署到 Linux 则是严重漏洞。

**修复建议**:
```python
SAFE_ROOT = Path("./md_docs").resolve()
path = Path(req.file_path).resolve()
if not str(path).startswith(str(SAFE_ROOT)):
    raise HTTPException(status_code=403, detail="禁止访问该路径")
```

---

### 🔴 S-03: 无任何认证/授权机制

**文件**: `src/server/app.py` 全部端点

`POST /api/v1/search`、`POST /api/v1/ingest`、`DELETE /api/v1/documents/{file_name}` 均无任何认证。任何能访问 `http://localhost:8000` 的人都能读取、写入、删除向量库数据。

**修复建议**: 至少添加一个静态 API Key 验证中间件：
```python
from fastapi import Header, HTTPException

async def verify_api_key(x_api_key: str = Header(None)):
    expected = os.getenv("MD_VECTOR_API_KEY", "")
    if expected and x_api_key != expected:
        raise HTTPException(status_code=401)
```

---

### 🟠 S-04: 无速率限制

`POST /api/v1/ingest` 和 `POST /api/v1/search` 无速率限制，每次请求都触发一次嵌入计算（CPU 密集型）。攻击者可以短时间内发送大量请求耗尽 CPU/内存。

**修复建议**: 引入 `slowapi` 或 `fastapi-limiter` 限制每 IP 的请求频率。

---

### 🟠 S-05: 错误信息泄露

**文件**: `src/server/app.py:128`

```python
except Exception as e:
    raise HTTPException(status_code=500, detail=str(e))
```

直接将异常信息返回给客户端，可能暴露文件路径、内部类名、库版本等敏感信息。

**修复建议**: 返回通用错误消息，详细错误记入日志：
```python
except Exception:
    logger.exception("ingest failed")
    raise HTTPException(status_code=500, detail="入库失败")
```

---

### 🟡 S-06: FastAPI 默认 CORS 无限制

**文件**: `src/server/app.py:80-85`

FastAPI 默认不设置 CORS 中间件，但在同源场景没问题。若后续被其他域名下的前端页面调用，需要显式配置 `CORSMiddleware`。

---

## 二、正确性与可靠性 (Correctness & Reliability)

### 🔴 C-01: ChromaDB 并发不安全

**文件**: `src/core/db.py`, `src/server/app.py`

`ChromaDB` 的 `PersistentClient` 使用 SQLite 作为底层存储，SQLite 在默认配置下不支持多写者并发。当前所有端点通过模块级单例共享同一个 `PersistentClient`，FastAPI 的线程池模型下多个请求并发写操作可能导致数据库锁错误甚至数据损坏。

**修复建议**: 
1. 使用 `chromadb.Client` (内存模式) + 定期持久化，或
2. 在写操作上加 `threading.Lock`，或
3. 升级到 ChromaDB 服务模式 (客户端-服务端架构)

---

### 🟠 C-02: Embedder 全局环境变量副作用

**文件**: `src/core/embedder.py:6-7`

```python
os.environ.setdefault("HF_HUB_OFFLINE", "1")
os.environ.setdefault("TRANSFORMERS_OFFLINE", "1")
```

模块导入时即修改全局环境变量，影响同一进程中的所有库。任何其他使用 HuggingFace 的代码都会受到意外影响。`__init__` 中还临时 `pop`/`set` 这些变量——在并发场景下会产生竞态条件。

**修复建议**: 不用全局环境变量，改为通过 `SentenceTransformer` 构造参数控制：
```python
self._model = SentenceTransformer(
    config.local_model,
    local_files_only=True,
)
```
下载回退逻辑单独提供一个 `download_model()` 脚本。

---

### 🟠 C-03: 分块 overlap 逻辑缺陷

**文件**: `src/core/ingest.py:113-123`

```python
if self.overlap > 0 and len(current) > self.overlap:
    current = current[-self.overlap:] + "\n\n" + para
else:
    current = para
```

当一个 chunk flush 后，`current` 被设为上一块的末尾 100 字符 + 新段落。但下一轮循环中 `len(current)` 可能已经接近 100 且小于 `max_size`，导致下一块的实际内容丰富度不足，overlap 区段被重复入库。**语义上问题不大（overlap 本就是设计如此），但 extra chunks 可能导致搜索结果中出现高度重复内容。**

---

### 🟠 C-04: `SearchRequest.top_k` 无上界

**文件**: `src/server/app.py:23-24`

```python
class SearchRequest(BaseModel):
    query: str
    top_k: int = 10
```

可传入 `top_k=999999` 导致 ChromaDB 尝试返回巨量数据，消耗内存。

**修复建议**:
```python
top_k: int = Field(default=10, ge=1, le=100)
```

---

### 🟠 C-05: 大文件入库无限制导致 OOM

**文件**: `src/core/ingest.py:183-216`

`ingest_content()` 将整个文件分块后一次性调用 `self.embedder.embed(texts)` 嵌入所有 chunks。对于超大文档（如 500KB+ 的 Markdown），生成的 chunks 数量和嵌入向量可能耗尽内存。

**修复建议**: 对 chunks 进行分批嵌入，每批最多 32 条：
```python
BATCH_SIZE = 32
for i in range(0, len(chunks), BATCH_SIZE):
    batch = chunks[i:i+BATCH_SIZE]
    ...
```

---

### 🟡 C-06: 配置路径硬编码

**文件**: 多处

`"config.yaml"` 字符串硬编码在 `config.py:62`, `server/app.py:47,59`, `cli/main.py:26` 共 4 处。修改文件名需要改动所有位置。

---

### 🟡 C-07: health 端点不检查实际依赖

**文件**: `src/server/app.py:94-96`

```python
@app.get("/api/v1/health")
def health():
    return {"status": "ok"}
```

无论 ChromaDB 是否可访问、模型是否加载成功，始终返回 `ok`。这不是真正的健康检查。

---

## 三、设计缺陷 (Design)

### 🟠 D-01: Embedder 违反单一职责原则

**文件**: `src/core/embedder.py`

`Embedder` 同时包含本地模型和 API 两种完全不同的嵌入实现，通过 `if config.mode == "local"` 分支判断。每增加一种嵌入后端都需要修改 `__init__`、`dimension`、`embed` 三个方法。

**修复建议**: 拆分为策略模式：
```python
class Embedder(Protocol):
    def embed(self, texts: list[str]) -> list[list[float]]: ...
    @property
    def dimension(self) -> int: ...

class LocalEmbedder: ...
class APIEmbedder: ...

def create_embedder(config: EmbedConfig) -> Embedder:  # 工厂
    if config.mode == "local": return LocalEmbedder(config)
    if config.mode == "api": return APIEmbedder(config)
```

---

### 🟠 D-02: 服务层模块级全局单例

**文件**: `src/server/app.py:28-31`

```python
_db: VectorDB | None = None
_searcher: Searcher | None = None
_ingestor: DocumentIngestor | None = None
```

模块级全局变量是经典的**反模式**：
- 无法在测试中替换依赖（必须靠环境变量 hack）
- 无法为不同请求使用不同配置
- FastAPI 的依赖注入被完全绕过

**修复建议**: 使用 FastAPI 的 `Depends` 注入：
```python
def get_searcher() -> Searcher:
    ...
    return searcher

@app.post("/api/v1/search")
def search_documents(req: SearchRequest, searcher: Searcher = Depends(get_searcher)):
    ...
```

---

### 🟠 D-03: 同步嵌入阻塞事件循环

**文件**: `src/server/app.py:132-134`, `src/core/embedder.py:55-61`

`embedder.embed()` 是同步方法（CPU 密集型），在 FastAPI 的 async 事件循环中调用会阻塞整个线程。FastAPI 默认线程池很小，多个并发搜索请求会排队等待。

**修复建议**: 使用 `run_in_executor` 将嵌入放到线程池：
```python
import asyncio
loop = asyncio.get_event_loop()
embeddings = await loop.run_in_executor(None, self.embedder.embed, texts)
```

---

### 🟡 D-04: `AppConfig` 双重初始化模式

**文件**: `src/core/config.py:44-57`

```python
@dataclass
class AppConfig:
    chroma: ChromaConfig = field(default_factory=ChromaConfig)  # dataclass 字段
    ...
    def __init__(self, **kwargs):  # 自定义 __init__ 覆盖 dataclass 的 __init__
        self.chroma = ChromaConfig(**kwargs.get("chroma", {}))
```

`@dataclass` 生成的 `__init__` 被完全替换，`field(default_factory=...)` 成为死代码但仍在类定义中（混淆阅读者）。`__eq__` 和 `__repr__` 仍由 dataclass 生成，行为与手写的 `__init__` 不一致（`AppConfig()` 的 `chroma` 字段在 `field()` 定义中是 `ChromaConfig()` 但在 `__init__` 中也是 `ChromaConfig()` ——刚好一致，但依赖巧合）。

---

### 🟡 D-05: `DocumentIngestor` 未测试核心流程

**文件**: `tests/test_ingest.py:72-87`

`TestDocumentIngestor` 只测了"读取 Markdown 文件"这一行代码，没有测试 `ingest_content()`（分块→嵌入→入库的核心路径）。真正的 `DocumentIngestor` 集成测试隐藏在 `test_search.py` 的 fixture 里。这导致修改 `ingest.py` 时可能没有直接的测试反馈。

---

## 四、可维护性 (Maintainability)

### 🟡 M-01: CLI 和 Server 重复组件初始化逻辑

**文件**: `src/cli/main.py:26-33` vs `src/server/app.py:34-74`

两处各自实现了一套几乎相同的 `_get_components` / `_get_db` + `_get_embedder` + `_init_services` 逻辑。如果 VectorDB 或 Embedder 的初始化签名改变，需要改两个地方。

**修复建议**: 在 `core` 层抽象一个 `AppContext` 工厂：
```python
# src/core/context.py
@dataclass
class AppContext:
    config: AppConfig
    db: VectorDB
    embedder: Embedder
    searcher: Searcher
    ingestor: DocumentIngestor

def create_context(config_path: str = "config.yaml") -> AppContext:
    ...
```

---

### 🟡 M-02: 无日志系统

整个项目无结构化日志。所有输出都通过 `typer.echo()` 或 `print()`。API 请求无日志记录，错误信息只能靠 FastAPI 的默认 stderr。

---

### 🟡 M-03: API 模型无请求校验

**文件**: `src/server/app.py:16-19`

```python
class IngestRequest(BaseModel):
    file_path: str | None = None
    content: str | None = None
    file_name: str | None = None
```

缺少对 `file_name` 的格式校验——可传入包含路径分隔符、空字符串或非法字符的文件名。

---

## 五、测试覆盖 (Testing)

### 测试覆盖概览

| 模块 | 测试数 | 覆盖关键路径? | 缺失 |
|------|--------|--------------|------|
| config | 5 | ✅ | - |
| db | 4 | ✅ | - |
| embedder | 5 | ⚠️ | API 模式未测、Invalid mode 未测 |
| ingest | 6 | ⚠️ | DocumentIngestor 核心未测 |
| search | 5 | ✅ | - |
| api | 6 | ⚠️ | DELETE 未测、ingest file_path 未测 |

### 🔴 T-01: API 嵌入模式零测试

**文件**: `tests/test_embedder.py`

全部 5 个测试仅测 `local` 模式。`Embedder(mode="api")` 的初始化、`dimension` 属性、`embed()` 方法均无测试覆盖。如果 API 模式的代码被重构引入 bug，不会触发任何测试失败。

### 🟠 T-02: DELETE 端点零测试

**文件**: `tests/test_api.py`

`DELETE /api/v1/documents/{file_name}` 无任何测试。删除不存在的文件、删除已入库的文件、删除后搜索验证——全部缺失。

### 🟡 T-03: 空数据搜索测试语义不准确

**文件**: `tests/test_api.py:53-61`

`test_search_empty_collection` 断言空 collection 搜索返回 `[]`。但 ChromaDB 在空 collection 下 `query()` 的行为可能随时间变化——当前"恰好"返回空，但这是未定义行为。

---

## 六、性能 (Performance)

### 🟡 P-01: `list_sources()` 全量扫描

**文件**: `src/core/search.py:65-74`

```python
def list_sources(self) -> list[str]:
    result = self.collection.get(include=["metadatas"])  # 取全部 metadatas
```

当 collection 有数十万条 chunks 时，一次性拉取全部 metadata 到内存做去重，极其浪费。

**修复建议**: 使用 ChromaDB 的 `get` 方法配合 `limit` 和分页，或将 source 列表单独维护在一个轻量集合中（如 Python set 持久化到 JSON 文件）。

---

### 🟡 P-02: 每次 CLI 命令重建 Embedder

**文件**: `src/cli/main.py:26-33`

每个 CLI 命令调用 `_get_components()` 都会重新创建 `VectorDB` + `Embedder`。模型加载（即使从缓存）也要 1-2 秒。对于连续执行的命令，这是不必要的开销。

---

## 七、汇总

### 按严重等级统计

| 等级 | 数量 | 条目 |
|------|------|------|
| 🔴 严重 | 4 | S-01(明文密钥), S-02(路径遍历), S-03(无认证), C-01(并发不安全) |
| 🟠 高 | 8 | S-04, S-05, C-02, C-03, C-04, C-05, D-01, D-02, D-03, T-02 |
| 🟡 中 | 11 | S-06, C-06, C-07, D-04, D-05, M-01, M-02, M-03, T-01, T-03, P-01, P-02 |
| 🔵 低 | 0 | - |

### 修复优先级

```
Phase 1 (立即)
├── 🔴 S-02: 路径遍历防护 → 加路径白名单
├── 🔴 S-03: 无认证 → 加 API Key 中间件
├── 🔴 C-01: 并发安全 → 加写锁
└── 🔴 S-01: 明文密钥 → 改为环境变量

Phase 2 (本周)
├── 🟠 C-04: top_k 无上界 → Pydantic Field(le=100)
├── 🟠 C-05: 大文件 OOM → 分批嵌入
├── 🟠 S-05: 错误信息泄露 → 通用错误 + 日志
├── 🟠 D-02: 全局单例 → FastAPI Depends 注入
└── 🟠 T-02: DELETE 测试缺失 → 补充

Phase 3 (迭代)
├── 🟡 D-01: Embedder 拆分 → 策略模式
├── 🟡 M-01: 重复逻辑 → AppContext 工厂
├── 🟡 M-02: 无日志 → logging 模块
├── 🟡 T-01: API 模式测试 → 补充
└── 🟡 P-01: list_sources 全量 → 分页
```