diff --git a/qy_lty/.planning/ROADMAP.md b/qy_lty/.planning/ROADMAP.md index b2600b7..5ef0e72 100644 --- a/qy_lty/.planning/ROADMAP.md +++ b/qy_lty/.planning/ROADMAP.md @@ -1,75 +1,77 @@ -# Roadmap：QY LTY Backend - -## 概览 - -本路线图聚焦 **Milestone v1.0「通用凭据槽位（APP ID + Access Token）」**：在后端落地一组全局单例的通用凭据存储槽位，让 Web 管理后台可写入、手机端 + 设备端可读取，且 Access Token 不会落入生产日志。粒度为 coarse（3-5 phase），按"数据层 → 管理端读写 → 客户端读取 + 日志脱敏"自下而上推进，三个 phase 串行依赖。 - -## Milestones - -- 🚧 **v1.0 通用凭据槽位** — Phase 1-3（启动 2026-05-07） - -## Phases - -**Phase 编号说明：** -- 整数 phase（1、2、3）：当期 milestone 计划工作 -- 小数 phase（2.1、2.2）：紧急插入工作（标记 INSERTED） - -小数 phase 在数值序内夹在前后整数之间执行。 - -- [x] **Phase 1: 凭据槽位数据层** — 落地 `CredentialSlot` 单例模型 + 迁移 + Django Admin 注册（脱敏 + 隐藏新增按钮）✓ 2026-05-07 完成（Plan 01-01 + 01-02） -- [x] **Phase 2: 管理端读写接口** — 在 `/api/v1/admin/` 暴露凭据槽位 GET（脱敏）/ PUT（覆写）端点，admin token 鉴权 ✓ 2026-05-07 完成（Plan 02-01 + 02-02） -- [ ] **Phase 3: 客户端读取与日志脱敏** — 在 `/api/credential-slot/` 暴露明文读取端点（user token 鉴权），并在阿里云日志链路过滤 `access_token` - -## Phase Details - -### Phase 1: 凭据槽位数据层 -**Goal**: 在数据库层落地全局单例的凭据槽位，并通过 Django Admin 提供受控录入入口（写入态可见、查看态脱敏、不可新增多条） -**Depends on**: Nothing（首个 phase） -**Requirements**: CRED-01, CRED-02 -**Success Criteria**（必须为真）: - 1. 在 Django shell / Admin 中尝试创建第二条 `CredentialSlot` 记录会被 DB 层或模型层拒绝（DB 中最多一条） - 2. 运行 `python manage.py migrate` 后，schema 中存在 `app_id`、`access_token`、`updated_at` 三个字段，且首次访问时通过 `get_or_create(pk=1)` 拿到一条空记录 - 3. 登录 Django Admin（SimpleUI 主题）打开凭据槽位页面：列表态/查看态下 `access_token` 显示为脱敏掩码（仅末 4 位），编辑态下显示明文供运营录入 - 4. Admin 列表页**不显示**「新增」按钮（强制单例语义，避免运营误建第二条） -**Plans:** 2 plans - - [x] 01-01-PLAN.md — 凭据槽位单例模型 + 迁移 + mask_token 工具（CRED-01）✓ 2026-05-07 完成（commits a9c25eb / 30c7caf / a475fe4） - - [x] 01-02-PLAN.md — Django Admin 注册（脱敏/单例新增/禁删）+ 修改记录两条（CRED-02）✓ 2026-05-07 完成（commits 653f057 / ddbcb7d；Task 2 checkpoint 由 orchestrator Django test client 程序化验收 10/10 PASS） - -### Phase 2: 管理端读写接口 -**Goal**: Web 管理后台（qy-lty-admin）能通过 `/api/v1/admin/credential-slot/` 读取脱敏后的凭据槽位、并以全字段覆写方式更新它 -**Depends on**: Phase 1 -**Requirements**: CRED-03, CRED-04 -**Success Criteria**（必须为真）: - 1. 携带有效 `admin_token:{token}` 调用 `GET /api/v1/admin/credential-slot/`，返回 `{ success, code, message, data: { app_id, access_token: , updated_at } }`，其中 `access_token` 仅暴露末 4 位掩码 - 2. 携带有效 `admin_token:{token}` 调用 `PUT /api/v1/admin/credential-slot/` 提交 `{ app_id, access_token }`，记录被全字段覆写、`updated_at` 自动刷新；空记录场景自动 `get_or_create`，不报 404 - 3. 不携带 admin token、或仅携带普通 user token 调用上述两个端点均被拒绝（401 / 403），错误响应同样符合 `StandardResponseMiddleware` 壳层 - 4. 接口出现在 `/swagger/` 与 `/redoc/` 中，请求/响应 schema 与实际行为一致（drf-yasg 自动生成） -**Plans:** 2 plans - - [x] 02-01-PLAN.md — CredentialSlot serializer + view（GET 脱敏 / PUT 覆写 + admin 二次校验）+ admin_urls 路由注册（CRED-03 + CRED-04）✓ 2026-05-07（commits 6820fe7 / 192d0a1 / 9d02021） - - [x] 02-02-PLAN.md — 端到端 curl + Django shell 验收 8 条 success criteria + qy_lty / qy-lty-admin 两端修改记录互引（CRED-03 + CRED-04）✓ 2026-05-07（commits 3cfd481 / 46d72b8） - -### Phase 3: 客户端读取与日志脱敏 -**Goal**: 手机端（LTY_App_Project_URP）和设备端（LTY_Project）能通过 `/api/credential-slot/` 拿到**明文** APP ID + Access Token 去调用第三方服务；同时确保 Access Token 在阿里云日志中始终脱敏，不论是 PUT 请求体还是管理端 GET 响应体 -**Depends on**: Phase 2 -**Requirements**: CRED-05, CRED-06 -**Success Criteria**（必须为真）: - 1. 携带有效 `token:{token}` 调用 `GET /api/credential-slot/`，返回 `{ success, code, message, data: { app_id, access_token: <明文>, updated_at } }`，Access Token 为明文（客户端实际调用第三方需要） - 2. 不携带 user token、或携带过期 token 调用客户端端点均被 `RedisTokenAuthentication` 拒绝（401），错误响应符合标准壳层 - 3. 在生产日志（阿里云日志服务）中检索 Phase 2 / Phase 3 的请求轨迹：`PUT /api/v1/admin/credential-slot/` 请求体里的 `access_token` 字段被脱敏；管理端 `GET` 响应体里的 `access_token` 同样脱敏；客户端明文 GET 端点的响应体不写入日志（或同样脱敏），无任何位置暴露完整 Access Token 明文 - 4. 端到端验证：管理后台用 PUT 写入一组凭据 → 手机端调用客户端 GET 拿到的 `app_id` / `access_token` 与管理端写入的一致（往返一致性成立） -**Plans**: TBD - -## Progress - -**执行顺序：** -Phase 按数值顺序执行：1 → 2 → 3（如出现紧急插入，记为 1.1 / 2.1 等） - -| Phase | Plans Complete | Status | Completed | -|-------|----------------|--------|-----------| -| 1. 凭据槽位数据层 | 2/2 | ✓ Complete | 2026-05-07 | -| 2. 管理端读写接口 | 2/2 | ✓ Complete | 2026-05-07 | -| 3. 客户端读取与日志脱敏 | 0/TBD | Not started | - | - ---- - -*生成时间：2026-05-07，Milestone v1.0「通用凭据槽位（APP ID + Access Token）」启动* +# Roadmap：QY LTY Backend + +## 概览 + +本路线图聚焦 **Milestone v1.0「通用凭据槽位（APP ID + Access Token）」**：在后端落地一组全局单例的通用凭据存储槽位，让 Web 管理后台可写入、手机端 + 设备端可读取，且 Access Token 不会落入生产日志。粒度为 coarse（3-5 phase），按"数据层 → 管理端读写 → 客户端读取 + 日志脱敏"自下而上推进，三个 phase 串行依赖。 + +## Milestones + +- 🚧 **v1.0 通用凭据槽位** — Phase 1-3（启动 2026-05-07） + +## Phases + +**Phase 编号说明：** +- 整数 phase（1、2、3）：当期 milestone 计划工作 +- 小数 phase（2.1、2.2）：紧急插入工作（标记 INSERTED） + +小数 phase 在数值序内夹在前后整数之间执行。 + +- [x] **Phase 1: 凭据槽位数据层** — 落地 `CredentialSlot` 单例模型 + 迁移 + Django Admin 注册（脱敏 + 隐藏新增按钮）✓ 2026-05-07 完成（Plan 01-01 + 01-02） +- [x] **Phase 2: 管理端读写接口** — 在 `/api/v1/admin/` 暴露凭据槽位 GET（脱敏）/ PUT（覆写）端点，admin token 鉴权 ✓ 2026-05-07 完成（Plan 02-01 + 02-02） +- [ ] **Phase 3: 客户端读取与日志脱敏** — 在 `/api/credential-slot/` 暴露明文读取端点（user token 鉴权），并在阿里云日志链路过滤 `access_token` + +## Phase Details + +### Phase 1: 凭据槽位数据层 +**Goal**: 在数据库层落地全局单例的凭据槽位，并通过 Django Admin 提供受控录入入口（写入态可见、查看态脱敏、不可新增多条） +**Depends on**: Nothing（首个 phase） +**Requirements**: CRED-01, CRED-02 +**Success Criteria**（必须为真）: + 1. 在 Django shell / Admin 中尝试创建第二条 `CredentialSlot` 记录会被 DB 层或模型层拒绝（DB 中最多一条） + 2. 运行 `python manage.py migrate` 后，schema 中存在 `app_id`、`access_token`、`updated_at` 三个字段，且首次访问时通过 `get_or_create(pk=1)` 拿到一条空记录 + 3. 登录 Django Admin（SimpleUI 主题）打开凭据槽位页面：列表态/查看态下 `access_token` 显示为脱敏掩码（仅末 4 位），编辑态下显示明文供运营录入 + 4. Admin 列表页**不显示**「新增」按钮（强制单例语义，避免运营误建第二条） +**Plans:** 2 plans + - [x] 01-01-PLAN.md — 凭据槽位单例模型 + 迁移 + mask_token 工具（CRED-01）✓ 2026-05-07 完成（commits a9c25eb / 30c7caf / a475fe4） + - [x] 01-02-PLAN.md — Django Admin 注册（脱敏/单例新增/禁删）+ 修改记录两条（CRED-02）✓ 2026-05-07 完成（commits 653f057 / ddbcb7d；Task 2 checkpoint 由 orchestrator Django test client 程序化验收 10/10 PASS） + +### Phase 2: 管理端读写接口 +**Goal**: Web 管理后台（qy-lty-admin）能通过 `/api/v1/admin/credential-slot/` 读取脱敏后的凭据槽位、并以全字段覆写方式更新它 +**Depends on**: Phase 1 +**Requirements**: CRED-03, CRED-04 +**Success Criteria**（必须为真）: + 1. 携带有效 `admin_token:{token}` 调用 `GET /api/v1/admin/credential-slot/`，返回 `{ success, code, message, data: { app_id, access_token: , updated_at } }`，其中 `access_token` 仅暴露末 4 位掩码 + 2. 携带有效 `admin_token:{token}` 调用 `PUT /api/v1/admin/credential-slot/` 提交 `{ app_id, access_token }`，记录被全字段覆写、`updated_at` 自动刷新；空记录场景自动 `get_or_create`，不报 404 + 3. 不携带 admin token、或仅携带普通 user token 调用上述两个端点均被拒绝（401 / 403），错误响应同样符合 `StandardResponseMiddleware` 壳层 + 4. 接口出现在 `/swagger/` 与 `/redoc/` 中，请求/响应 schema 与实际行为一致（drf-yasg 自动生成） +**Plans:** 2 plans + - [x] 02-01-PLAN.md — CredentialSlot serializer + view（GET 脱敏 / PUT 覆写 + admin 二次校验）+ admin_urls 路由注册（CRED-03 + CRED-04）✓ 2026-05-07（commits 6820fe7 / 192d0a1 / 9d02021） + - [x] 02-02-PLAN.md — 端到端 curl + Django shell 验收 8 条 success criteria + qy_lty / qy-lty-admin 两端修改记录互引（CRED-03 + CRED-04）✓ 2026-05-07（commits 3cfd481 / 46d72b8） + +### Phase 3: 客户端读取与日志脱敏 +**Goal**: 手机端（LTY_App_Project_URP）和设备端（LTY_Project）能通过 `/api/credential-slot/` 拿到**明文** APP ID + Access Token 去调用第三方服务；同时确保 Access Token 在阿里云日志中始终脱敏，不论是 PUT 请求体还是管理端 GET 响应体 +**Depends on**: Phase 2 +**Requirements**: CRED-05, CRED-06 +**Success Criteria**（必须为真）: + 1. 携带有效 `token:{token}` 调用 `GET /api/credential-slot/`，返回 `{ success, code, message, data: { app_id, access_token: <明文>, updated_at } }`，Access Token 为明文（客户端实际调用第三方需要） + 2. 不携带 user token、或携带过期 token 调用客户端端点均被 `RedisTokenAuthentication` 拒绝（401），错误响应符合标准壳层 + 3. 在生产日志（阿里云日志服务）中检索 Phase 2 / Phase 3 的请求轨迹：`PUT /api/v1/admin/credential-slot/` 请求体里的 `access_token` 字段被脱敏；管理端 `GET` 响应体里的 `access_token` 同样脱敏；客户端明文 GET 端点的响应体不写入日志（或同样脱敏），无任何位置暴露完整 Access Token 明文 + 4. 端到端验证：管理后台用 PUT 写入一组凭据 → 手机端调用客户端 GET 拿到的 `app_id` / `access_token` 与管理端写入的一致（往返一致性成立） +**Plans:** 2 plans + - [ ] 03-01-PLAN.md — 客户端凭据槽位 GET 接口（CRED-05：CredentialSlotClientView 明文返回 + /api/credential-slot/ 路由注册） + - [ ] 03-02-PLAN.md — 阿里云日志 access_token 脱敏（CRED-06：AccessTokenMaskFilter + LOGGING 配置 + 修改记录） + +## Progress + +**执行顺序：** +Phase 按数值顺序执行：1 → 2 → 3（如出现紧急插入，记为 1.1 / 2.1 等） + +| Phase | Plans Complete | Status | Completed | +|-------|----------------|--------|-----------| +| 1. 凭据槽位数据层 | 2/2 | ✓ Complete | 2026-05-07 | +| 2. 管理端读写接口 | 2/2 | ✓ Complete | 2026-05-07 | +| 3. 客户端读取与日志脱敏 | 0/2 | Not started | - | + +--- + +*生成时间：2026-05-07，Milestone v1.0「通用凭据槽位（APP ID + Access Token）」启动* diff --git a/qy_lty/.planning/phases/03-client-and-log-mask/03-01-PLAN.md b/qy_lty/.planning/phases/03-client-and-log-mask/03-01-PLAN.md new file mode 100644 index 0000000..dbdf416 --- /dev/null +++ b/qy_lty/.planning/phases/03-client-and-log-mask/03-01-PLAN.md @@ -0,0 +1,590 @@ +--- +phase: 03-client-and-log-mask +plan: 01 +type: execute +wave: 1 +depends_on: [] +files_modified: + - aiapp/views.py + - qy_lty/urls.py +autonomous: true +requirements: + - CRED-05 +must_haves: + truths: + - "持有效 user token 调用 GET /api/credential-slot/ 返回 200 + 标准壳层 + 明文 access_token" + - "持有效 admin token 调用同接口同样返回 200 + 明文（不区分 admin/user token）" + - "无 token 调用返回 401，标准壳层 success=false" + - "持伪造 / 过期 token 调用返回 401（RedisTokenAuthentication 拒绝）" + - "/swagger.json/ 含 /api/credential-slot/ 路径条目，GET 方法 description 标注「明文返回」" + artifacts: + - path: "aiapp/views.py" + provides: "CredentialSlotClientView APIView 类（仅 GET 明文）+ _credential_slot_client_data_schema swagger schema" + contains: "class CredentialSlotClientView" + - path: "qy_lty/urls.py" + provides: "/api/credential-slot/ 路由注册（顶层 api_urlpatterns）" + contains: "client_credential_slot" + key_links: + - from: "qy_lty/urls.py:api_urlpatterns" + to: "aiapp.views.CredentialSlotClientView" + via: "from aiapp.views import CredentialSlotClientView" + pattern: "path\\('credential-slot/', CredentialSlotClientView" + - from: "CredentialSlotClientView.get" + to: "CredentialSlot.get_solo() + CredentialSlotSerializer + success_response" + via: "1:1 复刻 CredentialSlotAdminView.get 但不调 _ensure_admin / _build_response_data / mask_token" + pattern: "success_response\\(data=serializer\\.data" +--- + + +落地 CRED-05：在 `/api/credential-slot/` 暴露**客户端**通用凭据槽位 GET 接口，user / admin token 鉴权（`RedisTokenAuthentication` + `IsAuthenticated`），**明文**返回 `{ app_id, access_token, updated_at }`，供手机端（LTY_App_Project_URP）与设备端（LTY_Project）实际调用第三方服务（阿里云 / 火山 / 腾讯）。 + +**Purpose**: Milestone v1.0「通用凭据槽位」Phase 3 第一阶段。Phase 2 已让管理端读 / 写脱敏后的槽位；本 plan 让客户端读到明文。view 行为与 Phase 2 admin view 严格反向（明文返回、不调 mask_token、不做 is_staff 二次校验），路由在顶层 `api_urlpatterns` 而非 `userapp/admin_urls.py`。 + +**Output**: +- `aiapp/views.py` 末尾新增 `CredentialSlotClientView` 类（仅 GET）+ `_credential_slot_client_data_schema` 客户端响应 schema +- `qy_lty/urls.py` 顶部 imports 追加 `from aiapp.views import CredentialSlotClientView`、`api_urlpatterns` 列表中追加路由 +- 新接口 `/api/credential-slot/` 通 swagger 自动暴露 + + + +@$HOME/.claude/get-shit-done/workflows/execute-plan.md +@$HOME/.claude/get-shit-done/templates/summary.md + + + +@.planning/PROJECT.md +@.planning/ROADMAP.md +@.planning/STATE.md +@.planning/phases/03-client-and-log-mask/03-CONTEXT.md +@.planning/phases/03-client-and-log-mask/03-RESEARCH.md +@.planning/phases/02-admin-rest/02-01-SUMMARY.md +@.planning/phases/02-admin-rest/02-02-SUMMARY.md + +# 1:1 模板源 + 直接复用件 +@aiapp/views.py +@aiapp/serializers.py +@aiapp/models.py +@common/utils.py +@common/responses.py +@common/swagger_utils.py +@userapp/authentication.py +@userapp/utils.py +@qy_lty/urls.py + + + + +From `aiapp/views.py:1-19`（imports 段，**全部已就位**，本 plan 不需要再加任何 import）： +```python +from rest_framework.views import APIView +from rest_framework import status +from .models import ChatMessage, Bot, CredentialSlot # CredentialSlot 已 import +from .serializers import ChatMessageSerializer, CredentialSlotSerializer # 已 import +from rest_framework.permissions import IsAuthenticated # 已 import +from userapp.authentication import RedisTokenAuthentication # 已 import +from common.swagger_utils import swagger_schema, get_standardized_response_schema # 已 import +from common.responses import success_response, created_response, error_response # 已 import +from common.utils import mask_token # 已 import（本 plan 不调用，但 import 已存在无影响） +from drf_yasg import openapi # 已 import +from drf_yasg.utils import swagger_auto_schema # 已 import +import logging +logger = logging.getLogger(__name__) +``` + +From `aiapp/models.py`（CredentialSlot 单例契约，Phase 1 已交付）： +```python +class CredentialSlot(models.Model): + app_id = models.CharField(max_length=128, blank=True, default='') + access_token = models.CharField(max_length=512, blank=True, default='') + updated_at = models.DateTimeField(auto_now=True) + + @classmethod + def get_solo(cls): + # get_or_create(pk=1) 单一入口 + ... +``` + +From `aiapp/serializers.py:11-30`（Phase 2 已交付，**直接复用**，不新建客户端专用 serializer）： +```python +class CredentialSlotSerializer(serializers.ModelSerializer): + class Meta: + model = CredentialSlot + fields = ['app_id', 'access_token', 'updated_at'] + read_only_fields = ['updated_at'] +``` + +From `common/responses.py`：`success_response(data=None, message="操作成功", code=200)` → DRF Response 200 + `{success, code, message, data}` 标准壳层（StandardResponseMiddleware 不会二次包装）。 + +From `userapp/authentication.py:14-34` + `userapp/utils.py:39-45`：`RedisTokenAuthentication` 双查 `admin_token:{token}` → `token:{token}`，admin / user 都返回 `is_authenticated=True` 的 ParadiseUser → `IsAuthenticated` 对两种 token 都通过。**本 view 不调 is_staff 二次校验**（CONTEXT 锁定决策）。 + +From `aiapp/views.py:600-687`（Phase 2 模板，**1:1 复刻 GET 部分**，删 `_ensure_admin` / `_build_response_data` / PUT 三处）。 + +From `qy_lty/urls.py:48-60`（当前 api_urlpatterns 实测，本 plan 在 `common/upload/` 之后、`v1/admin/` 之前插入新行）： +```python +api_urlpatterns = [ + path('user/', include('userapp.urls')), + path('ai/', include('aiapp.urls')), + path('device/', include('device_interaction.urls')), + path('card/', include('card.urls')), + path('achievement/', include('achievement_app.urls')), + path('food/', include('food_app.urls')), + path('common/upload/', upload_file, name='file-upload'), + # ↓ 在此处插入新行 ↓ + path('v1/admin/', include('userapp.admin_urls')), +] +``` + + + + + + + Task 1: 在 aiapp/views.py 末尾追加 CredentialSlotClientView + 客户端响应 schema + aiapp/views.py + + + 1. **完整读 `aiapp/views.py:560-687`** —— Phase 2 `CredentialSlotPutRequestSchema` + `_credential_slot_data_schema` + `CredentialSlotAdminView` 完整段，作为 1:1 模板（删 PUT、删 `_ensure_admin`、删 `_build_response_data` / `mask_token` 调用） + 2. **完整读 `aiapp/views.py:1-19`** —— imports 段，确认 `CredentialSlot` / `CredentialSlotSerializer` / `RedisTokenAuthentication` / `IsAuthenticated` / `success_response` / `get_standardized_response_schema` / `openapi` / `swagger_auto_schema` 全部已 import，**本 task 不要再加任何 import** + 3. 读 `aiapp/serializers.py:11-30` 确认 `CredentialSlotSerializer` 字段集 = `['app_id', 'access_token', 'updated_at']`，明文存储（脱敏由 view 层负责，admin 层调 mask_token，client 层不调） + 4. 读 `common/responses.py:41-52` 确认 `success_response(data=..., message="...")` 调用契约 + 5. 读 `userapp/authentication.py` + `userapp/utils.py:39-45` 确认 admin/user token 都通过 `IsAuthenticated` + + + +**位置**：`aiapp/views.py` 末尾（紧邻 `CredentialSlotAdminView` 类的最后一行 `return success_response(data=data, message="凭据已更新")` 之后），追加以下 **两块代码**： + +**第 1 块：客户端响应 data 子 schema**（紧邻文件末尾，放在新 view 类**之前**，与 `_credential_slot_data_schema` 形成对称命名）： + +```python +# ====================================================================== +# Phase 3 — 通用凭据槽位客户端读取接口（CRED-05） +# 1:1 复刻 CredentialSlotAdminView 的 GET 部分，删 _ensure_admin / _build_response_data / PUT +# 关键差异：明文返回 access_token，不调 mask_token，不做 is_staff 二次校验 +# ====================================================================== + +# 客户端响应 data 子 schema：access_token 字段 description 显式标注「明文」 +_credential_slot_client_data_schema = openapi.Schema( + type=openapi.TYPE_OBJECT, + properties={ + 'app_id': openapi.Schema( + type=openapi.TYPE_STRING, + description='第三方服务商分配的 APP ID（明文）', + ), + 'access_token': openapi.Schema( + type=openapi.TYPE_STRING, + description='明文 Access Token，供手机/设备端实际调用第三方服务（管理端同接口会脱敏返回末 4 位）', + ), + 'updated_at': openapi.Schema( + type=openapi.TYPE_STRING, + format='date-time', + description='最近一次更新时间（ISO 8601）', + ), + }, +) +``` + +**第 2 块：CredentialSlotClientView 类**（紧接上面 schema 之后）： + +```python +class CredentialSlotClientView(APIView): + """通用凭据槽位客户端读取接口（user / admin token 鉴权，明文返回）。 + + GET: 返回明文 app_id + access_token，供手机/设备端实际调用第三方服务。 + + 与管理端 CredentialSlotAdminView 的关键差异（per CONTEXT.md D-Client-View）： + - 删 _ensure_admin：不做 is_staff 二次校验，admin / user token 都允许（admin 用户是手机用户超集） + - 删 _build_response_data：直接返回 serializer.data（不调 mask_token，明文返回） + - 删 PUT 方法：客户端只读，不能写入 + """ + authentication_classes = [RedisTokenAuthentication] + permission_classes = [IsAuthenticated] + tags = ['通用凭据槽位（客户端）'] + + @swagger_auto_schema( + operation_description="读取通用凭据槽位（明文 access_token，供手机/设备端实际调用第三方服务）", + responses={ + 200: openapi.Response( + '读取成功', + get_standardized_response_schema(_credential_slot_client_data_schema), + ), + 401: openapi.Response('未提供有效 token', get_standardized_response_schema()), + }, + security=[{'Bearer': []}], + tags=['通用凭据槽位（客户端）'], + ) + def get(self, request): + instance = CredentialSlot.get_solo() + serializer = CredentialSlotSerializer(instance) + return success_response(data=serializer.data, message="读取成功") +``` + +**严格禁止做的事**： +- ✗ 不要新增任何 import（imports 段已全部就位） +- ✗ 不要新建 `CredentialSlotClientSerializer`（CONTEXT 锁定复用 Phase 2 serializer） +- ✗ 不要在 view 内调 `mask_token`（CONTEXT 锁定明文返回） +- ✗ 不要在 view 内做 `request.user.is_staff` 校验（CONTEXT 锁定 admin/user 都允许） +- ✗ 不要新增 PUT method（客户端只读） +- ✗ 不要在 view 内打 `logger.info(serializer.data)` 或 `logger.info(request.data)` 类型语句（避免 access_token 进日志；plan 03-02 的 filter 会兜底，但 plan 03-01 不引入泄露源） +- ✗ 不要修改 Phase 2 的 `CredentialSlotAdminView` / `_credential_slot_data_schema` / `CredentialSlotPutRequestSchema`（Phase 2 已验收 PASS，本 plan 仅追加，不动既有） + + + + +# 1. 文件含新类 + schema（不再用 _ensure_admin 计数断言：admin view 实际含 1 处方法定义 + 2 处调用 = 3+ 处， +# 而客户端 view 类体内 0 处的判断由下面 #2 的 AST-equivalent 检查保证） +python -c "src=open('aiapp/views.py',encoding='utf-8').read(); assert 'class CredentialSlotClientView(APIView):' in src, '缺 CredentialSlotClientView 类'; assert '_credential_slot_client_data_schema' in src, '缺客户端 schema'; print('OK: view 类 + schema 已落地')" + +# 2. 客户端 view 类体不调 mask_token / _ensure_admin / _build_response_data / PUT +# 注意：生产 Docker 是 Python 3.8（ast.unparse 是 3.9+ 新增），所以用 re.search 切类体 + grep 校验， +# AST-equivalent 但兼容 3.8。 +python -c "import re, pathlib; src=pathlib.Path('aiapp/views.py').read_text(encoding='utf-8'); m=re.search(r'^class CredentialSlotClientView.*?(?=^class |\Z)', src, re.S | re.M); assert m, 'CredentialSlotClientView 未找到'; body=m.group(0); [None for forbidden in ['_ensure_admin', '_build_response_data', 'def put', 'mask_token'] if (lambda f: (_ for _ in ()).throw(AssertionError(f'客户端 view 类体不应含 {f}')) if f in body else None)(forbidden)]; print('OK: 客户端 view 类体不含 _ensure_admin / _build_response_data / def put / mask_token')" + +# 3. Django 系统检查 + URLConf 加载（在 plan 03-01 完成 task 2 后才会通过；本 task 仅校验 import 不报错） +python manage.py check aiapp + + + + + - aiapp/views.py 文件大小相比 Phase 2 收尾态增加 ~40-60 行 + - grep `class CredentialSlotClientView` 命中 1 次 + - grep `_credential_slot_client_data_schema` 命中 ≥ 2 次（schema 定义 + swagger_auto_schema 引用） + - 客户端 view 类体中 `mask_token` / `_ensure_admin` / `def put` / `_build_response_data` 0 命中 + - 客户端 view 类体中 `success_response(data=serializer.data` 命中 1 次（不是 `data=data`） + - `python manage.py check aiapp` 0 errors / 0 warnings + + + + `CredentialSlotClientView` 类与 `_credential_slot_client_data_schema` 已追加到 `aiapp/views.py` 末尾；imports 未变；Phase 2 既有代码未动；`python manage.py check aiapp` 通过。 + + + + + Task 2: 在 qy_lty/urls.py 注册 /api/credential-slot/ 路由 + import CredentialSlotClientView + qy_lty/urls.py + + + 1. **完整读 `qy_lty/urls.py:1-83`** —— 确认现有 imports 段（行 17-26）+ `api_urlpatterns` 列表（行 49-60）+ 顶层 `urlpatterns`（行 63-73） + 2. 注意行 26 已有 `from common.views import upload_file`，本 task 在该行**之后**追加 `from aiapp.views import CredentialSlotClientView` + 3. 注意行 57 是 `path('common/upload/', upload_file, name='file-upload'),`，行 59 是 `path('v1/admin/', include('userapp.admin_urls')),`，本 task 在这两行**之间**插入新路由 + 4. 读 `aiapp/views.py` 末尾（Task 1 完成后），确认 `CredentialSlotClientView` 已存在 + + + +**修改 1：顶部 imports 段追加** + +`qy_lty/urls.py` 行 26（`from common.views import upload_file` 之后）追加 1 行： + +```python +from aiapp.views import CredentialSlotClientView +``` + +**修改 2：api_urlpatterns 列表中追加路由** + +`qy_lty/urls.py` 行 57-59 段，在 `path('common/upload/', upload_file, name='file-upload'),` **之后**、`path('v1/admin/', include('userapp.admin_urls')),` **之前**追加 1 行（保持「客户端散点路径 → admin 命名空间」的视觉分组）： + +修改前（行 49-60，verbatim）： +```python +api_urlpatterns = [ + path('user/', include('userapp.urls')), + path('ai/', include('aiapp.urls')), + # path('ali/vi/api/', include('ali_vi_app.urls')), + path('device/', include('device_interaction.urls')), + path('card/', include('card.urls')), + path('achievement/', include('achievement_app.urls')), # 成就系统API + path('food/', include('food_app.urls')), # 食物管理API + path('common/upload/', upload_file, name='file-upload'), + # 管理员API接口路径（v1版本） + path('v1/admin/', include('userapp.admin_urls')), +] +``` + +修改后： +```python +api_urlpatterns = [ + path('user/', include('userapp.urls')), + path('ai/', include('aiapp.urls')), + # path('ali/vi/api/', include('ali_vi_app.urls')), + path('device/', include('device_interaction.urls')), + path('card/', include('card.urls')), + path('achievement/', include('achievement_app.urls')), # 成就系统API + path('food/', include('food_app.urls')), # 食物管理API + path('common/upload/', upload_file, name='file-upload'), + # Phase 3 — 客户端通用凭据槽位读取接口（CRED-05，明文返回） + path('credential-slot/', CredentialSlotClientView.as_view(), name='client_credential_slot'), + # 管理员API接口路径（v1版本） + path('v1/admin/', include('userapp.admin_urls')), +] +``` + +**严格禁止做的事**： +- ✗ 不要把路由放进 `aiapp/urls.py`（CONTEXT 锁定路径 `/api/credential-slot/`，不要 `/api/ai/credential-slot/`） +- ✗ 不要放进 `userapp/admin_urls.py`（那是 admin 命名空间，给 Phase 2 用） +- ✗ 不要改 `path('v1/admin/', include('userapp.admin_urls'))` 等任何既有行 +- ✗ 不要改 `urlpatterns`（行 63-73）—— 那是顶层包装，不是新增点 + + + + +# 1. import + path 行存在 +python -c "src=open('qy_lty/urls.py',encoding='utf-8').read(); assert 'from aiapp.views import CredentialSlotClientView' in src, '缺 CredentialSlotClientView import'; assert \"path('credential-slot/', CredentialSlotClientView.as_view(), name='client_credential_slot')\" in src, '缺路由注册'; print('OK: imports + path 已落地')" + +# 2. URL 解析正确（应返回 CredentialSlotClientView） +python manage.py shell -c "from django.urls import resolve; m = resolve('/api/credential-slot/'); print('OK: resolved to', m.func.view_class.__name__); assert m.func.view_class.__name__ == 'CredentialSlotClientView'; assert m.url_name == 'client_credential_slot'" + +# 3. 反向解析 +python manage.py shell -c "from django.urls import reverse; u = reverse('client_credential_slot'); print('OK: reverse =', u); assert u == '/api/credential-slot/'" + +# 4. Django 系统检查全通过 +python manage.py check + + + + + - `qy_lty/urls.py` 第 27 行附近含 `from aiapp.views import CredentialSlotClientView` + - `api_urlpatterns` 列表的 `common/upload/` 与 `v1/admin/` 之间含新行 `path('credential-slot/', CredentialSlotClientView.as_view(), name='client_credential_slot')` + - `resolve('/api/credential-slot/')` 返回 `CredentialSlotClientView` 视图，url_name = `client_credential_slot` + - `reverse('client_credential_slot')` 返回 `/api/credential-slot/` + - `python manage.py check` 0 errors / 0 warnings + - `aiapp/urls.py` / `userapp/admin_urls.py` 完全未动 + + + + `/api/credential-slot/` 已注册到顶层 `api_urlpatterns`，指向 `CredentialSlotClientView`；URL 解析与反向解析均通过；`python manage.py check` 全绿。 + + + + + Task 3: 启动验证 + Django test client 端到端验收（CRED-05 五条 truth） + _phase3_01_verify.py + + + 1. 完整读 `.planning/phases/02-admin-rest/02-02-SUMMARY.md`（或 `02-VERIFICATION.md`），确认 Phase 2 已落地的 DB 探针态：`pk=1` 的 `CredentialSlot.app_id='probe_app'` / `access_token='probe_secret_xxxx'` + 2. 读 `userapp/utils.py:generate_token` —— Redis key 体系：`admin_token:{token}` / `token:{token}`，TTL 30 天 + 3. 读 Phase 2 验收脚本风格（`_phase2_verify.py` 已删，可参考 `02-VERIFICATION.md` 中保留的命令清单） + 4. 读 `qy_lty/settings.py:LOGGING` —— 客户端 logger 走 `aiapp` / `userapp` / `common` 三个 logger，level=INFO + + + +在仓库根目录创建 `_phase3_01_verify.py`（**临时验收脚本，phase end 由 plan 03-02 负责删除**），内容如下： + +```python +"""Phase 3 Plan 01 验收脚本（CRED-05 客户端 GET 端到端 + Swagger）。 + +验收 6 条 truth： + T1: 持 user token GET → 200 + 标准壳层 + 明文 access_token + T2: 持 admin token GET → 200 + 明文（不区分） + T3: 无 token GET → 401 + 标准壳层 success=false + T4: 持伪造 token GET → 401 + T5: /swagger.json/ 含 /credential-slot/（注：basePath=/api，所以 schema paths key 不带 /api 前缀） + T6: 响应 access_token 与 DB instance.access_token 字符级一致（明文，不脱敏） + +执行： + cd + python manage.py shell < _phase3_01_verify.py + +验收完毕由 Plan 03-02 Task 4 删除本文件 + 还原 DB 探针态（保持与 Phase 2 探针一致）。 +""" +import json +import secrets +import django + +django.setup() if not django.apps.apps.ready else None + +from django.test import Client +from django.core.cache import cache +from aiapp.models import CredentialSlot +from userapp.models import ParadiseUser + +PASS = [] +FAIL = [] + +def assert_eq(name, got, expected): + if got == expected: + PASS.append(f'{name}: got={got!r}') + else: + FAIL.append(f'{name}: expected={expected!r} got={got!r}') + +def assert_in(name, needle, haystack): + if needle in haystack: + PASS.append(f'{name}: {needle!r} in haystack') + else: + FAIL.append(f'{name}: {needle!r} NOT in haystack={haystack!r}') + +# 0) 准备：DB 探针 + 两个 token（admin / user） +slot, _ = CredentialSlot.objects.get_or_create(pk=1) +slot.app_id = 'probe_app' +slot.access_token = 'probe_secret_xxxx' +slot.save() + +# 拿一个 staff user + 一个普通 user +staff = ParadiseUser.objects.filter(is_staff=True).first() +normal = ParadiseUser.objects.filter(is_staff=False).first() +if not staff or not normal: + FAIL.append('环境缺 staff/normal user，请先 createsuperuser 或注册一个 normal user') +else: + admin_token = secrets.token_hex(18) + user_token = secrets.token_hex(18) + cache.set(f'admin_token:{admin_token}', staff.id, timeout=300) + cache.set(f'token:{user_token}', normal.id, timeout=300) + + client = Client() + + # T1: user token → 200 + 明文 + r = client.get('/api/credential-slot/', HTTP_AUTHORIZATION=f'Bearer {user_token}') + assert_eq('T1.status_code', r.status_code, 200) + body = r.json() + assert_eq('T1.success', body.get('success'), True) + assert_eq('T1.data.app_id', body['data'].get('app_id'), 'probe_app') + assert_eq('T1.data.access_token (plaintext)', body['data'].get('access_token'), 'probe_secret_xxxx') + # 明文断言 —— 必须等于原 DB 值，不能含 * + if '*' in (body['data'].get('access_token') or ''): + FAIL.append('T1.no_mask: 客户端响应不应含 *（应明文返回）') + else: + PASS.append('T1.no_mask: 客户端响应未脱敏 ✓') + + # T2: admin token → 200 + 明文（不区分） + r = client.get('/api/credential-slot/', HTTP_AUTHORIZATION=f'Bearer {admin_token}') + assert_eq('T2.status_code', r.status_code, 200) + body = r.json() + assert_eq('T2.data.access_token (plaintext)', body['data'].get('access_token'), 'probe_secret_xxxx') + + # T3: 无 token → 401 + r = client.get('/api/credential-slot/') + assert_eq('T3.status_code', r.status_code, 401) + body = r.json() + assert_eq('T3.success', body.get('success'), False) + + # T4: 伪造 token → 401 + r = client.get('/api/credential-slot/', HTTP_AUTHORIZATION='Bearer fake_token_zzz_not_in_redis') + assert_eq('T4.status_code', r.status_code, 401) + + # T5: swagger 暴露 + r = client.get('/swagger.json/') + assert_eq('T5.swagger.status_code', r.status_code, 200) + schema = r.json() + # StandardResponseMiddleware 也会包 OpenAPI schema → unwrap data 字段 + if 'paths' not in schema and 'data' in schema: + schema = schema['data'] + paths = schema.get('paths', {}) + assert_in('T5.swagger.paths', '/credential-slot/', list(paths.keys())) + if '/credential-slot/' in paths: + ops = paths['/credential-slot/'] + assert_in('T5.swagger.has_get', 'get', ops) + + # 还原 + 清理 + slot.refresh_from_db() + assert_eq('T6.db_unchanged.app_id', slot.app_id, 'probe_app') + assert_eq('T6.db_unchanged.access_token', slot.access_token, 'probe_secret_xxxx') + cache.delete(f'admin_token:{admin_token}') + cache.delete(f'token:{user_token}') + +print('=' * 70) +print(f'PASS ({len(PASS)}):') +for p in PASS: + print(f' ✓ {p}') +if FAIL: + print(f'FAIL ({len(FAIL)}):') + for f in FAIL: + print(f' ✗ {f}') + raise SystemExit(1) +else: + print('ALL PASS') +``` + +**执行**： +```bash +python manage.py shell < _phase3_01_verify.py +``` + +**严格禁止做的事**： +- ✗ 不要在脚本里写明文 user-token / admin-token 进 git（用 `secrets.token_hex(18)` 即时生成，验完 cache.delete） +- ✗ 不要改 DB 探针的最终值（验收前后 `app_id='probe_app'` / `access_token='probe_secret_xxxx'` 保持不变） +- ✗ 不要把脚本提交到 git（Plan 03-02 Task 4 会负责删除；可加 `_phase3_*.py` 到 `.gitignore` 或直接不 add） +- ✗ 不要让脚本启 daphne / runserver（用 `django.test.Client` in-process） + + + + +# 跑端到端脚本 +python manage.py shell < _phase3_01_verify.py + +# 脚本本身的退出码即验收结果（FAIL 时 exit 1） + + + + + - 脚本运行最后输出 `ALL PASS` + - PASS 行数 ≥ 12（5 条 truth × 平均 2-3 个独立断言） + - FAIL 行数 = 0 + - DB 状态在脚本结束后保持 `app_id='probe_app'` / `access_token='probe_secret_xxxx'` 探针态 + - Redis cache 中临时 token key 已清理（cache.delete 调用过） + - 脚本**不**进 git（保留在 working tree 供 Plan 03-02 Task 4 删除） + + + + `_phase3_01_verify.py` 输出 `ALL PASS`，CRED-05 五条 success criteria 全部通过；DB 还原；临时 token 清理。 + + + + + + +## Trust Boundaries + +| Boundary | Description | +|----------|-------------| +| 客户端 → API | Unity 设备 / 手机端通过 HTTP + Bearer token 调用 `/api/credential-slot/`；token 明文走 Authorization header | +| API → Redis | `RedisTokenAuthentication` 双查 `admin_token:{token}` / `token:{token}` 的 user_id | +| API → DB | `CredentialSlot.get_solo()` 查 PostgreSQL 单例记录 | + +## STRIDE Threat Register + +| Threat ID | Category | Component | Disposition | Mitigation Plan | +|-----------|----------|-----------|-------------|-----------------| +| T-03-01 | Spoofing | `RedisTokenAuthentication` | mitigate | 复用 Phase 1/2 已验收的 Bearer token 双查机制；token 由 `userapp.utils.generate_token` 生成 30 天 TTL，伪造概率 = 1/256^36 | +| T-03-02 | Information Disclosure | 客户端 GET 响应明文 access_token | accept | CONTEXT 已论证：客户端必须拿到明文才能调第三方服务；HTTPS 由 Nginx 反代兜底（CLAUDE.md 部署段）；不在 Phase 3 范围 | +| T-03-03 | Information Disclosure | 客户端 GET 响应明文 access_token 经 logger 打到生产日志 | mitigate | **plan 03-02 引入 `AccessTokenMaskFilter`** 在 LOGGING.handlers 层兜底；本 plan 不引入新 logger.info 调用，确保 plan 03-01 不创造新泄露源 | +| T-03-04 | Elevation of Privilege | 普通 user token 持有者通过 client view 读取明文 access_token | accept | CONTEXT 已论证：admin user 是手机用户超集；明文 access_token 是「第三方服务的口令」非「平台用户的凭证」，user 持有它仅能调阿里云 / 火山等第三方 API（这正是设计意图） | +| T-03-05 | Tampering | 攻击者修改路由让 client view 走入 admin 路径 | mitigate | 路由放 `qy_lty/urls.py:api_urlpatterns` 顶层（非 sub-include），不被任何 app urls.py 覆盖 | +| T-03-06 | Repudiation | 客户端调用未留审计日志 | accept | 当前架构无审计需求（候选优先级 #5 待 pytest 体系落地后再做） | +| T-03-07 | DoS | 高频 GET 调用 | accept | 候选优先级未列「客户端调用频率限流」；Nginx 反代可手动加 rate limit；不在 Phase 3 范围 | + + + +**Plan 整体验收（汇总 task 1/2/3 的自动化检查）**： + +```bash +# 1. 静态：view + schema 已追加 +python -c "src=open('aiapp/views.py',encoding='utf-8').read(); assert 'class CredentialSlotClientView(APIView):' in src; assert '_credential_slot_client_data_schema' in src; print('OK: view + schema')" + +# 2. 静态：路由已注册 +python -c "src=open('qy_lty/urls.py',encoding='utf-8').read(); assert 'from aiapp.views import CredentialSlotClientView' in src; assert \"path('credential-slot/', CredentialSlotClientView.as_view()\" in src; print('OK: route')" + +# 3. 动态：Django check + URL 解析 +python manage.py check +python manage.py shell -c "from django.urls import resolve; assert resolve('/api/credential-slot/').func.view_class.__name__ == 'CredentialSlotClientView'; print('OK: resolve')" + +# 4. 动态：端到端 5 条 truth +python manage.py shell < _phase3_01_verify.py +``` + + + +- [ ] `aiapp/views.py` 末尾追加 `CredentialSlotClientView` 类与 `_credential_slot_client_data_schema`，imports 段未变 +- [ ] `qy_lty/urls.py` imports 段追加 `from aiapp.views import CredentialSlotClientView`，`api_urlpatterns` 列表中含 `path('credential-slot/', CredentialSlotClientView.as_view(), name='client_credential_slot')` +- [ ] `python manage.py check` 0 errors / 0 warnings +- [ ] `resolve('/api/credential-slot/')` 返回 `CredentialSlotClientView` +- [ ] `_phase3_01_verify.py` 输出 `ALL PASS`，5 条 truth（user/admin token 200 + 明文、无 token 401、伪造 token 401、swagger 暴露）全过 +- [ ] DB 探针态 `pk=1 / app_id='probe_app' / access_token='probe_secret_xxxx'` 保持 +- [ ] `aiapp/urls.py` / `userapp/admin_urls.py` / `aiapp/serializers.py` / `aiapp/models.py` 等其他文件**未改动** +- [ ] Phase 2 既有 `CredentialSlotAdminView` 行为未变（Phase 2 验收脚本若重跑应仍 PASS — 本 plan 不影响） + + + +完成后创建 `.planning/phases/03-client-and-log-mask/03-01-SUMMARY.md`，记录： +- 实际改动文件 + 行号区间 +- `_phase3_01_verify.py` 输出截图（PASS 列表） +- 与 Phase 2 admin view 的对照差异（删 `_ensure_admin` / 删 `_build_response_data` / 删 PUT / 不调 mask_token） +- 留给 Plan 03-02 的 hand-off：DB 探针态保持、`_phase3_01_verify.py` 待 Plan 03-02 Task 4 删除 + + diff --git a/qy_lty/.planning/phases/03-client-and-log-mask/03-02-PLAN.md b/qy_lty/.planning/phases/03-client-and-log-mask/03-02-PLAN.md new file mode 100644 index 0000000..bd2954a --- /dev/null +++ b/qy_lty/.planning/phases/03-client-and-log-mask/03-02-PLAN.md @@ -0,0 +1,991 @@ +--- +phase: 03-client-and-log-mask +plan: 02 +type: execute +wave: 2 +depends_on: + - 03-01 +files_modified: + - common/logging/__init__.py + - common/logging/filters.py + - qy_lty/settings.py + - docs/修改记录.md +autonomous: true +requirements: + - CRED-06 +must_haves: + truths: + - "AccessTokenMaskFilter 类存在于 common/logging/filters.py，是 logging.Filter 子类" + - "LOGGING.filters 中注册 access_token_mask；LOGGING.handlers.aliyun 与 LOGGING.handlers.console 各引用 access_token_mask" + - "filter 处理伪 LogRecord（msg 含 access_token 明文）后 record.getMessage() 不含完整明文，含末 4 位脱敏掩码" + - "filter 在 4 种序列化形态（JSON / Python dict repr / URL query / 等号或冒号兜底）下均能脱敏" + - "filter 不误伤 Authorization header / Bearer token 等非 access_token 字段" + - "Django 启动 + LOGGING dictConfig 加载无 ValueError（filter 工厂语法 () 写正确）" + - "docs/修改记录.md 顶部追加 [2026-05-08] Phase 3 条目，覆盖 client view + filter + LOGGING 三处改动" + artifacts: + - path: "common/logging/__init__.py" + provides: "package marker（空文件，让 common.logging 成为可 import 的包）" + - path: "common/logging/filters.py" + provides: "AccessTokenMaskFilter(logging.Filter) + 4 个 regex 模式 + filter() 方法" + contains: "class AccessTokenMaskFilter" + - path: "qy_lty/settings.py" + provides: "LOGGING.filters 段 + handlers.aliyun.filters + handlers.console.filters" + contains: "access_token_mask" + - path: "docs/修改记录.md" + provides: "顶部追加 [2026-05-08] Phase 3 条目" + contains: "[2026-05-08] Phase 3" + key_links: + - from: "qy_lty/settings.py:LOGGING.filters" + to: "common.logging.filters.AccessTokenMaskFilter" + via: "() 工厂语法（dictConfig 标准）" + pattern: "common\\.logging\\.filters\\.AccessTokenMaskFilter" + - from: "LOGGING.handlers.aliyun" + to: "filter 实例" + via: "filters: ['access_token_mask']" + pattern: "filters.*access_token_mask" + - from: "LOGGING.handlers.console" + to: "filter 实例" + via: "filters: ['access_token_mask']" + pattern: "filters.*access_token_mask" + - from: "AccessTokenMaskFilter._mask_in_text" + to: "common.utils.mask_token" + via: "from common.utils import mask_token" + pattern: "mask_token\\(" +--- + + +落地 CRED-06：在 Python `logging` 链路新增 `AccessTokenMaskFilter`，挂到 LOGGING.handlers.aliyun / console 上，确保**任何**未来开发者写 `logger.info(f"PUT body: {request.data}")` 类型代码时 `access_token` 字段值会被自动脱敏（保留末 4 位），覆盖 4 种序列化形态：JSON 字符串 / Python dict repr / URL query / 等号或冒号兜底。 + +**Purpose**: Milestone v1.0「通用凭据槽位」Phase 3 第二阶段（防御性兜底）。RESEARCH 已实证当前仓库**没有**任何代码 logger 输出 `CredentialSlot.access_token` 明文（StandardResponseMiddleware 不打日志、view 不显式 logger 字段），所以 CRED-06 的端到端验证靠**单元测试**伪造 LogRecord 验证 filter 行为，不靠端到端找泄露路径。这是 CRED-06 的真实价值 —— 防御性兜底。 + +**Output**: +- `common/logging/__init__.py`（**新建**空文件） +- `common/logging/filters.py`（**新建**，含 `AccessTokenMaskFilter` 类 + 4 个 regex + `filter()` 方法） +- `qy_lty/settings.py`：`LOGGING` 字典追加 `filters` 段；`handlers.aliyun` / `handlers.console` 各加 `'filters': ['access_token_mask']` +- `docs/修改记录.md` 顶部追加 `[2026-05-08] Phase 3 — 客户端凭据槽位 GET 接口 + 阿里云日志脱敏` 条目 +- 临时验收脚本 `_phase3_02_verify.py` 跑完即删 + 还原 DB 探针 + 删除 plan 03-01 的 `_phase3_01_verify.py` + + + +@$HOME/.claude/get-shit-done/workflows/execute-plan.md +@$HOME/.claude/get-shit-done/templates/summary.md + + + +@.planning/PROJECT.md +@.planning/ROADMAP.md +@.planning/STATE.md +@.planning/phases/03-client-and-log-mask/03-CONTEXT.md +@.planning/phases/03-client-and-log-mask/03-RESEARCH.md +@.planning/phases/03-client-and-log-mask/03-01-SUMMARY.md + +# 直接复用件 +@common/utils.py +@common/aliyun_logging.py +@common/middleware.py +@qy_lty/settings.py +@CLAUDE.md +@docs/修改记录.md + + + + +From `common/utils.py:10-32`（**直接复用**，filter 内调用）： +```python +def mask_token(token: str, visible_tail: int = 4, mask_char: str = '*') -> str: + if not token: + return '' + if len(token) <= visible_tail: + return mask_char * len(token) + return mask_char * (len(token) - visible_tail) + token[-visible_tail:] +``` + +`mask_token('abcdefgh1234', visible_tail=4)` → `'********1234'`（**8** 颗 `*` + `1234`，共 12 字符；输入 12 字符长度保持）。 + +From `common/aliyun_logging.py:16-30`（emit 用 `record.getMessage()`，所以 filter 改 `record.msg` + `record.args` 后 emit 自然拿到脱敏后的字符串）： +```python +class AliyunLogHandler(logging.Handler): + def emit(self, record): + log_item = LogItem() + log_item.set_time(record.created) + log_item.set_contents([ + ('level', record.levelname), + ('message', record.getMessage()), # ← 此处实时拼出脱敏后字符串 + ... + ]) +``` + +From `qy_lty/settings.py:372-412`（当前 LOGGING 全貌，本 plan 改动点已在 RESEARCH Example 2 给出 diff）。 + +From `docs/修改记录.md` 头部「修改格式说明」段（**严格遵守的格式约束**）： +``` +### [日期] 修改简述 + +- **文件路径**: 相对于项目根目录的文件路径 +- **修改类型**: 新增 / 修改 / 删除 / 重构 / 修复Bug +- **修改内容**: 具体修改了什么 +- **修改原因**: 为什么要做这个修改 +``` +新条目追加在「修改历史」段最顶部（最新在最前），紧邻 `[2026-05-07] Phase 2` 条目**之上**。 + +From CLAUDE.md「项目修改记录规则」： +- `qy_lty/docs/修改记录.md` 仅记录服务端改动；qy-lty-admin 各自维护 +- 跨项目联动：两端各写一条，相互引用；本 phase **无跨项目联动**（CONTEXT 显式声明）—— Unity 客户端在 LTY_Project / LTY_App_Project_URP 独立 repo，不在 qy-lty-admin 范畴 + + + + + + + Task 1: 新建 common/logging/ 包目录 + AccessTokenMaskFilter 类（4 regex + filter 方法） + common/logging/__init__.py, common/logging/filters.py, _phase3_02_unit_test.py + + + 1. **完整读 `common/utils.py:10-32`** —— `mask_token` 行为契约（空输入返回 ''、短于 4 全脱、其余末 4 位明文）。注意 `mask_token('abcdefgh1234')` 输出 `'********1234'`（8 星 + 4 位），共 12 字符 + 2. 读 `common/aliyun_logging.py:16-30` —— `AliyunLogHandler.emit` 用 `record.getMessage()`，确认 filter 改 `record.msg` 后 emit 自然生效 + 3. 读 RESEARCH Pattern 3（`AccessTokenMaskFilter` 完整骨架，4 个 regex + filter 方法）+ Pitfall 1-5（filter 挂错位置 / dictConfig 工厂语法等） + 4. 读 RESEARCH 「附录：access_token 实际泄露路径详细分析」段 —— 确认本 filter 是「防御性兜底」，不是「修补现有泄露」 + 5. 验证 `common/` 是 utility 命名空间（已读 `common/utils.py:1-7` 注释「不是 Django app」），新建 `common/logging/` 子包不破坏此约定 + 6. **检查现有目录**：`ls common/` 应不含 `logging/` 子目录（确认本 task 是纯新建） + + + + +**新建文件 1**：`common/logging/__init__.py`（**空文件**，让 `common.logging` 成为可 import 的 Python 包）。 + +**严格要求**：内容**完全为空**（0 字节或仅 1 个换行）。不要写任何代码、注释、`__all__`、版本号 —— 任何内容都可能在未来引入意外副作用。 + +**新建文件 2**：`common/logging/filters.py`，**完整**写入以下内容（不能简化、不能省略 docstring、不能删 4 个 regex 中任何一个）： + +```python +"""通用日志脱敏 Filter 集合。 + +本模块挂载到 settings.LOGGING.filters，由 LOGGING.handlers.{aliyun|console} 引用， +覆盖所有 logger → handler 路径，对 record 内 access_token 字段值脱敏（保留末 4 位）。 + +设计动机（per CONTEXT.md / RESEARCH.md）： +- 当前仓库代码没有 logger 输出 CredentialSlot.access_token 明文的路径（已实证） +- 但 Phase 1 + Phase 2 + Phase 3 的 view 已让 access_token 进入「内存中可被随手 dump」的状态 +- 任何后续开发者写 logger.info(f"PUT body: {request.data}") 类型代码就会泄露 +- 本 filter 是「防御性兜底」：在 handler 层面统一兜住，不依赖每个 view 自律 +""" +import logging +import re + +from common.utils import mask_token + + +class AccessTokenMaskFilter(logging.Filter): + """识别日志记录中的 access_token 明文值并脱敏（保留末 4 位）。 + + 覆盖 4 种序列化形态： + 1. JSON 字符串（双引号）： '"access_token": "VALUE"' + 2. Python dict repr（单引号）："'access_token': 'VALUE'" + 3. URL query： 'access_token=VALUE&...' + 4. 兜底（等号或冒号 + 空格）： 'access_token: VALUE' / 'access_token = VALUE' + + 挂载点： + settings.LOGGING.filters['access_token_mask'] + → settings.LOGGING.handlers.{aliyun|console}.filters + + 设计要点： + - 仅识别 access_token 字段名为前缀锚点；不脱敏裸 token / Bearer / Authorization header + （那是另一类敏感数据，留 v2.x 候选优先级 #1 / #3 处理；详见 RESEARCH Pitfall 3） + - 同时改 record.msg 与 record.args，避免 Formatter 阶段再用 % 拼接出明文 + （详见 RESEARCH Pitfall 2） + - filter() 永远 return True，不丢弃 record（详见 RESEARCH Pitfall 1） + """ + + # 4 种序列化形态对应的正则模式 + _PATTERNS = ( + # 1) JSON 字符串：双引号；group 顺序 (前缀, 值, 后缀) + re.compile(r'("access_token"\s*:\s*")([^"]+)(")'), + # 2) Python dict repr：单引号；group 顺序 (前缀, 值, 后缀) + re.compile(r"('access_token'\s*:\s*')([^']+)(')"), + # 3) URL query：以 & / 空格 / 引号结尾；group 顺序 (前缀, 值) + re.compile(r'(access_token=)([^&\s"\']+)'), + # 4) 兜底：等号 / 冒号 + 可选空格；group 顺序 (前缀, 值) + # 用 [^\s,;)\]\}"\']+ 作为终止符以避免吃到下一个字段 + re.compile(r'(access_token\s*[:=]\s*)([^\s,;)\]\}"\']+)'), + ) + + # 快速短路：record.msg / args 中没有 access_token 字面量时直接返回，避免 4 次正则扫描 + _NEEDLE = 'access_token' + + def _sub(self, match: 're.Match') -> str: + """根据 group 数（2 或 3）调用 mask_token 重组匹配段。""" + groups = match.groups() + if len(groups) == 3: + # 模式 1 / 2：('"access_token":"', VALUE, '"') + return groups[0] + mask_token(groups[1]) + groups[2] + if len(groups) == 2: + # 模式 3 / 4：('access_token=', VALUE) + return groups[0] + mask_token(groups[1]) + return match.group(0) # 防御：未来若加新模式 group 数变了，原样返回避免崩溃 + + def _mask_in_text(self, text): + """对单个字符串依次应用 4 个正则；非字符串原样返回。""" + if not isinstance(text, str): + return text + if self._NEEDLE not in text.lower(): + return text + for pattern in self._PATTERNS: + text = pattern.sub(self._sub, text) + return text + + def filter(self, record: logging.LogRecord) -> bool: + # 1. record.msg 字符串脱敏（最常见的 logger.info("xxx %s", x) 之 "xxx %s" 部分；以及 logger.info(plain_str) 整串） + if isinstance(record.msg, str): + record.msg = self._mask_in_text(record.msg) + + # 2. record.args 中的元素脱敏 + # - dict 形态（logger.info("k=%(access_token)s", {'access_token': '...'})）：按 key 名直接脱敏值 + # - tuple 形态（logger.info("token=%s", token_str)）：按字符串内容脱敏 + if record.args: + if isinstance(record.args, dict): + new_args = {} + for k, v in record.args.items(): + if k == 'access_token' and isinstance(v, str): + new_args[k] = mask_token(v) + elif isinstance(v, str): + new_args[k] = self._mask_in_text(v) + else: + new_args[k] = v + record.args = new_args + elif isinstance(record.args, tuple): + record.args = tuple( + self._mask_in_text(a) if isinstance(a, str) else a + for a in record.args + ) + + # 永远不丢弃 record；filter 仅做改写 + return True +``` + +**新建文件 3（临时单元测试，verify 用，跑完不入 git）**：`_phase3_02_unit_test.py`（落地到仓库根，让 `python _phase3_02_unit_test.py` 直接跑；避免在 PowerShell 下用多行 `python -c` 引号转义易碎）： + +```python +"""Phase 3 Plan 02 Task 1 单元测试（filter 行为）。 + +跑： + python _phase3_02_unit_test.py + +验： + 1. import 链 OK + 继承 logging.Filter + 2. _PATTERNS 长度 = 4 + 3. 4 种序列化形态（JSON / Pyrepr / Query / Fallback）下伪 LogRecord 处理后含末 4 位、不含完整明文 + 4. 不误伤 Authorization header / Bearer 字段 + 5. tuple 形态 record.args 脱敏 + 6. 短 token（< 4 字符）全脱，不暴露长度 + 7. filter() 永远 return True + +跑完即删（由 Plan 03-02 Task 3 / Task 4 负责清理）。 +""" +import logging + +from common.logging.filters import AccessTokenMaskFilter + +# 1. 继承关系 +assert AccessTokenMaskFilter.__bases__[0].__name__ == 'Filter', '应继承 logging.Filter' +print('OK: import + 继承 logging.Filter') + +# 2. 4 模式 +assert len(AccessTokenMaskFilter._PATTERNS) == 4, f'_PATTERNS 长度应 = 4，实际 {len(AccessTokenMaskFilter._PATTERNS)}' +print('OK: 4 patterns') + +f = AccessTokenMaskFilter() + +# 3. 4 种形态 —— 用宽松断言（与 Plan 02 Task 3 T6 风格一致），避开 mask_token 长度计算细节 +fixtures = [ + ('JSON', '{"access_token": "abcdefgh1234"}'), + ('Pyrepr', "{'access_token': 'abcdefgh1234'}"), + ('Query', 'GET /x?access_token=abcdefgh1234&u=1'), + ('Fallback', 'access_token: abcdefgh1234'), +] +for name, msg in fixtures: + rec = logging.LogRecord('t', logging.INFO, '', 0, msg, None, None) + ret = f.filter(rec) + assert ret is True, f'{name}: filter() 应永远返回 True，实际 {ret!r}' + out = rec.getMessage() + assert '1234' in out, f'{name}: 末 4 位丢失 - {out!r}' + assert 'abcdefgh' not in out, f'{name}: 明文未脱敏 - {out!r}' + print(f'OK [{name}]: {out}') + +# 4. 不误伤 Authorization header / Bearer +for msg in ['Authorization header: bearer_user_token_xxxxxxx', 'Bearer raw_token_zzz']: + rec = logging.LogRecord('t', logging.INFO, '', 0, msg, None, None) + f.filter(rec) + out = rec.getMessage() + assert out == msg, f'误伤了非 access_token 字段：原={msg!r} 出={out!r}' + print(f'OK 不误伤: {msg}') + +# 5. tuple 形态 record.args 脱敏（logger.info('access_token=%s', token_str)） +rec = logging.LogRecord('t', logging.INFO, '', 0, 'access_token=%s', ('abcdefgh1234',), None) +f.filter(rec) +out = rec.getMessage() +assert 'abcdefgh' not in out, f'tuple args 未脱敏: {out!r}' +print(f'OK tuple args: {out}') + +# 6. 短 token 全脱 +rec = logging.LogRecord('t', logging.INFO, '', 0, '{"access_token": "abc"}', None, None) +f.filter(rec) +out = rec.getMessage() +assert 'abc' not in out, f'短 token 未脱敏: {out!r}' +assert '***' in out, f'短 token 应全部用 * 替换: {out!r}' +print(f'OK 短 token: {out}') + +print('=' * 60) +print('ALL UNIT TESTS PASS') +``` + +**严格禁止做的事**： +- ✗ 不要把 `mask_token` 改成局部 lambda（保持 import 复用 Phase 1 的实现，避免双源漂移） +- ✗ 不要在 filter 里用 `print()` 调试（filter 在 logging 链路内，print 会触发递归） +- ✗ 不要让 filter 返回 `False`（False 会丢弃 record，破坏其它日志输出） +- ✗ 不要用 `re.sub` 时把 `mask_token` 直接传入 —— 必须通过 `self._sub` 拿 `match.groups()` 处理 +- ✗ 不要把 4 个 regex 合并成 1 个大 regex（可读性 + group 数变量化会变脆） +- ✗ 不要把模式扩展到 `Authorization` / `Bearer` / `token` 裸字段（违反 Pitfall 3） +- ✗ 不要让 `_NEEDLE` 大小写敏感失效（用户大概率写 `'access_token'` 小写；若需大小写不敏感可未来迭代） +- ✗ `_phase3_02_unit_test.py` 不要 git add（验完由 Plan 03-02 Task 3 / Task 4 清理） + + + + +# 1. 文件存在 + 是包 +python -c "import os; assert os.path.isfile('common/logging/__init__.py'); assert os.path.isfile('common/logging/filters.py'); print('OK: package files exist')" + +# 2. __init__.py 是空文件（≤ 2 字节，允许结尾换行） +python -c "import os; sz = os.path.getsize('common/logging/__init__.py'); assert sz <= 2, f'__init__.py 应为空文件，实际 {sz} 字节'; print('OK: __init__.py empty')" + +# 3. 跑临时单元测试脚本（覆盖：import + 继承、4 模式、4 形态脱敏、不误伤、tuple args、短 token、filter return True） +# 用临时脚本而非多行 python -c，避免 PowerShell 嵌套引号转义问题 +python _phase3_02_unit_test.py + +# 4. settings 没在本 task 改动（Task 2 才改）—— 防御性快速 grep +python -c "src=open('qy_lty/settings.py',encoding='utf-8').read(); assert 'access_token_mask' not in src, 'Task 1 不应改 settings.py，那是 Task 2 的工作'; print('OK: Task 1 不动 settings.py')" + + + + + - `common/logging/__init__.py` 存在且为空文件（≤ 2 字节） + - `common/logging/filters.py` 存在，含 `class AccessTokenMaskFilter(logging.Filter)` + - `from common.logging.filters import AccessTokenMaskFilter` 不报错 + - `AccessTokenMaskFilter._PATTERNS` 长度 = 4 + - `python _phase3_02_unit_test.py` 输出 `ALL UNIT TESTS PASS` + - 4 种形态（JSON / Pyrepr / Query / Fallback）下伪 LogRecord 处理后 `record.getMessage()` 含 `1234` 末 4 位、不含完整明文 `abcdefgh` + - `Authorization header:` 和 `Bearer` 等非 access_token 字段值未被脱敏 + - tuple 形态 `record.args` 中的 access_token 值被脱敏 + - 短于 4 字符的 token 全部脱敏（不暴露长度） + - filter 返回 `True`（不丢弃 record） + - `qy_lty/settings.py` 未被本 task 改动（settings 改动归 Task 2） + + + + `common/logging/__init__.py` 与 `common/logging/filters.py` 已创建；`_phase3_02_unit_test.py` 输出 `ALL UNIT TESTS PASS`，覆盖 import / 4 模式 / 4 形态 / 不误伤 / tuple args / 短 token / filter() return True。 + + + + + Task 2: 在 qy_lty/settings.py 的 LOGGING 字典注册 filter（新增 filters 段 + handlers.aliyun/console 各加 filters 引用） + qy_lty/settings.py + + + 1. **完整读 `qy_lty/settings.py:370-412`** —— 当前 LOGGING 字典全貌，确认无 `filters` 段，handlers 仅 `aliyun` / `console` 两个，loggers 有 5 条 + 2. 读 RESEARCH Example 2 —— 完整 diff（修改前 / 修改后）+ Pitfall 5（dictConfig 工厂语法 `()` 而非 `class`） + 3. 读 RESEARCH Pitfall 1 —— filter 必须挂在 handlers 段，**不**挂在 loggers 段 + 4. 确认 `setup_logging()` 在 LOGGING 字典 dictConfig **之前**调用（`settings.py:371`）—— `setup_logging` 直接 addHandler 给 root logger，不走 dictConfig，本 task 改动不影响它 + + + + +**修改 `qy_lty/settings.py`**，定位在第 372-412 行的 `LOGGING = {...}` 块。改动有 3 处。 + +**修改前**（verbatim，行 372-412 当前实测）： +```python +LOGGING = { + 'version': 1, + 'disable_existing_loggers': False, + 'handlers': { + 'aliyun': { + 'level': 'INFO', + 'class': 'common.aliyun_logging.AliyunLogHandler', + }, + 'console': { + 'level': 'DEBUG', + 'class': 'logging.StreamHandler', + }, + }, + 'loggers': { + 'django': { + 'handlers': ['aliyun', 'console'], + 'level': 'INFO', + 'propagate': True, + }, + 'django.request': { + 'handlers': ['console'], + 'level': 'ERROR', + 'propagate': False, + }, + 'aiapp': { + 'handlers': ['aliyun', 'console'], + 'level': 'INFO', + 'propagate': True, + }, + 'common': { + 'handlers': ['aliyun', 'console'], + 'level': 'INFO', + 'propagate': True, + }, + 'userapp': { + 'handlers': ['aliyun', 'console'], + 'level': 'INFO', + 'propagate': True, + }, + }, +} +``` + +**修改后**（**整段替换**为以下内容，3 处 diff：① 在 `'disable_existing_loggers': False,` 与 `'handlers': {` 之间插入 `'filters'` 段；② `'aliyun'` handler 加 `'filters': ['access_token_mask'],`；③ `'console'` handler 加 `'filters': ['access_token_mask'],`；loggers 段**完全不动**）： + +```python +LOGGING = { + 'version': 1, + 'disable_existing_loggers': False, + # Phase 3 — Access Token 日志脱敏 filter（CRED-06） + # 挂载策略：filter 注册在 LOGGING.filters，再由 LOGGING.handlers 引用； + # 不挂在 loggers 段（per RESEARCH Pitfall 1：挂 logger 仅过滤直接通过该 logger 的 record， + # 挂 handler 才统一覆盖所有 logger → handler 路径） + 'filters': { + 'access_token_mask': { + '()': 'common.logging.filters.AccessTokenMaskFilter', + }, + }, + 'handlers': { + 'aliyun': { + 'level': 'INFO', + 'class': 'common.aliyun_logging.AliyunLogHandler', + 'filters': ['access_token_mask'], + }, + 'console': { + 'level': 'DEBUG', + 'class': 'logging.StreamHandler', + 'filters': ['access_token_mask'], + }, + }, + 'loggers': { + 'django': { + 'handlers': ['aliyun', 'console'], + 'level': 'INFO', + 'propagate': True, + }, + 'django.request': { + 'handlers': ['console'], + 'level': 'ERROR', + 'propagate': False, + }, + 'aiapp': { + 'handlers': ['aliyun', 'console'], + 'level': 'INFO', + 'propagate': True, + }, + 'common': { + 'handlers': ['aliyun', 'console'], + 'level': 'INFO', + 'propagate': True, + }, + 'userapp': { + 'handlers': ['aliyun', 'console'], + 'level': 'INFO', + 'propagate': True, + }, + }, +} +``` + +**严格禁止做的事**： +- ✗ 不要在 filters 段写 `'class': 'common.logging.filters.AccessTokenMaskFilter'` —— 必须用 `'()': '...'`（dictConfig 标准；handler 用 `'class'` 但 filter 用 `'()'`，两者语法不互通） +- ✗ 不要在 loggers 段加 filters 引用（违反 Pitfall 1） +- ✗ 不要去掉 `setup_logging()` 调用（行 371，那是阿里云 root handler 的独立路径，与 LOGGING dictConfig 互补） +- ✗ 不要改 handlers 的 `level` / `class` 字段 +- ✗ 不要改 loggers 的任何字段（5 条 logger 完全保持原样） +- ✗ 不要把 `disable_existing_loggers` 改成 `True` + + + + +# 1. settings.py 文件含新内容 +python -c "src=open('qy_lty/settings.py',encoding='utf-8').read(); assert \"'access_token_mask'\" in src; assert \"'()': 'common.logging.filters.AccessTokenMaskFilter'\" in src; assert src.count(\"['access_token_mask']\") >= 2, 'aliyun + console 两个 handler 都要挂 filter'; print('OK: settings 改动落地')" + +# 2. Django 启动 + LOGGING dictConfig 加载无 ValueError +python -c "import django; import os; os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'qy_lty.settings'); django.setup(); print('OK: Django 启动，LOGGING dictConfig 加载成功')" + +# 3. 实际拿到 filter 实例（验证 dictConfig 工厂语法 () 写正确） +python -c " +import django, os +os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'qy_lty.settings') +django.setup() +import logging +aliyun_handler = next(h for h in logging.getLogger('aiapp').handlers + logging.getLogger().handlers if 'AliyunLogHandler' in type(h).__name__) +console_handler = next(h for h in logging.getLogger('aiapp').handlers if 'StreamHandler' in type(h).__name__ and 'Aliyun' not in type(h).__name__) +for name, h in [('aliyun', aliyun_handler), ('console', console_handler)]: + fnames = [type(f).__name__ for f in h.filters] + assert 'AccessTokenMaskFilter' in fnames, f'{name} handler 缺 AccessTokenMaskFilter (got {fnames})' + print(f'OK [{name}] filters: {fnames}') +" + +# 4. 端到端：在 aiapp logger 上 logger.info('access_token=secret_zzz') 后 console 输出脱敏 +python -c " +import django, os, io, contextlib, logging +os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'qy_lty.settings') +django.setup() +log = logging.getLogger('aiapp') +buf = io.StringIO() +# 找 console handler 并临时把 stream 接到 buf +console = next(h for h in log.handlers if 'StreamHandler' in type(h).__name__ and 'Aliyun' not in type(h).__name__) +orig_stream = console.stream +console.stream = buf +try: + log.info('access_token=secret_zzz_ABCD') +finally: + console.stream = orig_stream +out = buf.getvalue() +assert 'secret_zzz_' not in out, f'明文泄露到 stderr: {out!r}' +assert 'ABCD' in out, f'末 4 位丢失: {out!r}' +print(f'OK 端到端：{out.strip()}') +" + +# 5. 既有 logger 配置未动 +python -c "src=open('qy_lty/settings.py',encoding='utf-8').read(); assert src.count(\"'handlers': ['aliyun', 'console']\") == 4, '应有 4 条 logger 用 [aliyun, console] handler 列表'; assert \"'level': 'INFO'\" in src; print('OK: loggers 段保持原状')" + + + + + - `qy_lty/settings.py` 的 `LOGGING` 字典含 `'filters'` 段，且 `'access_token_mask'` 用 `'()': 'common.logging.filters.AccessTokenMaskFilter'` 工厂语法 + - `LOGGING.handlers.aliyun.filters` = `['access_token_mask']` + - `LOGGING.handlers.console.filters` = `['access_token_mask']` + - `loggers` 段 5 条 logger 完全未动 + - `django.setup()` 不报 `ValueError: Unable to configure filter ...` + - `logging.getLogger('aiapp').handlers` 中两个 handler 的 `filters` 列表均含 `AccessTokenMaskFilter` 实例 + - 端到端：`logger.info('access_token=secret_zzz_ABCD')` 后 console 输出含 `ABCD` 不含 `secret_zzz_` + + + + LOGGING dictConfig 已扩展 filters 段并由 aliyun/console handler 引用；Django 启动无 ValueError；filter 实例已挂载，端到端输出验证脱敏生效。 + + + + + Task 3: 端到端验收脚本（CRED-05 + CRED-06 整合验证 + 还原 DB + 删除临时文件） + _phase3_02_verify.py + + + 1. 完整读 `_phase3_01_verify.py`（plan 03-01 创建的临时脚本，仍在 working tree） + 2. 读 `.planning/phases/02-admin-rest/02-VERIFICATION.md`（如存在）—— 参考端到端脚本的「证据落地 → 脚本删除」流程 + 3. 读 `userapp/utils.py:generate_token` —— Redis cache key 体系 + 4. 读 `aiapp/views.py:600-687`（`CredentialSlotAdminView`）—— 端到端 PUT roundtrip 测试需要走 admin token + + + + +在仓库根创建 `_phase3_02_verify.py`，跑完即删（**本 task 包含删除步骤**）。脚本验收 9 条 success criteria（覆盖 Plan 03-01 的 5 条 + Plan 03-02 的 3 条 filter 单元测试 + 1 条端到端 logger 真实输出）： + +```python +"""Phase 3 Plan 02 整合验收脚本（CRED-05 + CRED-06 端到端）。 + +9 条 truth： + T1: client GET 携 user token → 200 + 明文（覆盖 plan 03-01 已验项目） + T2: client GET 携 admin token → 200 + 明文（不区分） + T3: client GET 无 token → 401 + T4: client GET 伪造 token → 401 + T5: /swagger.json/ 含 /credential-slot/ 路径 + T6: AccessTokenMaskFilter 对 4 种形态（JSON / Pyrepr / Query / Fallback）伪 LogRecord 全脱敏 + T7: AccessTokenMaskFilter 不误伤 Authorization header / Bearer 字段 + T8: 端到端 admin PUT roundtrip → client GET 拿到一致明文（往返一致） + T9: 端到端 logger.info 真打印一条含 access_token 的消息 → console 输出脱敏（防御性兜底真实生效） + +执行： + python manage.py shell < _phase3_02_verify.py +""" +import io +import json +import logging +import secrets +import django + +django.setup() if not django.apps.apps.ready else None + +from django.test import Client +from django.core.cache import cache +from aiapp.models import CredentialSlot +from userapp.models import ParadiseUser + +PASS, FAIL = [], [] + +def ok(name, cond, hint=''): + (PASS if cond else FAIL).append(f'{name}{": " + hint if hint else ""}') + +def reset_probe(): + """还原 DB 探针态：app_id=probe_app, access_token=probe_secret_xxxx""" + slot, _ = CredentialSlot.objects.get_or_create(pk=1) + slot.app_id, slot.access_token = 'probe_app', 'probe_secret_xxxx' + slot.save() + return slot + +# 准备：DB 探针 + 两个 token +slot = reset_probe() +staff = ParadiseUser.objects.filter(is_staff=True).first() +normal = ParadiseUser.objects.filter(is_staff=False).first() +if not staff or not normal: + FAIL.append('环境缺 staff / normal user') + +admin_token = secrets.token_hex(18) +user_token = secrets.token_hex(18) +cache.set(f'admin_token:{admin_token}', staff.id, timeout=600) +cache.set(f'token:{user_token}', normal.id, timeout=600) + +c = Client() + +# T1: user token → 200 + 明文 +r = c.get('/api/credential-slot/', HTTP_AUTHORIZATION=f'Bearer {user_token}') +b = r.json() +ok('T1.user_token_200', r.status_code == 200, f'sc={r.status_code}') +ok('T1.success_true', b.get('success') is True) +ok('T1.app_id', b['data'].get('app_id') == 'probe_app') +ok('T1.access_token_plain', b['data'].get('access_token') == 'probe_secret_xxxx', 'should be plaintext') +ok('T1.no_mask_in_response', '*' not in (b['data'].get('access_token') or '')) + +# T2: admin token → 200 + 明文（不区分） +r = c.get('/api/credential-slot/', HTTP_AUTHORIZATION=f'Bearer {admin_token}') +b = r.json() +ok('T2.admin_token_200', r.status_code == 200) +ok('T2.access_token_plain', b['data'].get('access_token') == 'probe_secret_xxxx') + +# T3: 无 token → 401 +r = c.get('/api/credential-slot/') +ok('T3.no_token_401', r.status_code == 401) +ok('T3.success_false', r.json().get('success') is False) + +# T4: 伪造 token → 401 +r = c.get('/api/credential-slot/', HTTP_AUTHORIZATION='Bearer fake_token_zzz_NOT_IN_REDIS') +ok('T4.fake_token_401', r.status_code == 401) + +# T5: swagger 暴露 +r = c.get('/swagger.json/') +ok('T5.swagger_200', r.status_code == 200) +schema = r.json() +if 'paths' not in schema and 'data' in schema: + schema = schema['data'] +paths = schema.get('paths', {}) +ok('T5.path_in_schema', '/credential-slot/' in paths, f'keys={list(paths.keys())[:5]}...') +if '/credential-slot/' in paths: + ops = paths['/credential-slot/'] + ok('T5.has_get', 'get' in ops) + +# T6: filter 4 种形态（伪 LogRecord）—— 用宽松断言 +from common.logging.filters import AccessTokenMaskFilter +f = AccessTokenMaskFilter() +fixtures = { + 'JSON': '{"access_token": "abcdefgh1234"}', + 'Pyrepr': "{'access_token': 'abcdefgh1234'}", + 'Query': 'GET /x?access_token=abcdefgh1234&u=1', + 'Fallback': 'access_token: abcdefgh1234', +} +for name, msg in fixtures.items(): + rec = logging.LogRecord('t', logging.INFO, '', 0, msg, None, None) + f.filter(rec) + out = rec.getMessage() + ok(f'T6.{name}.no_plain', 'abcdefgh' not in out, f'out={out!r}') + ok(f'T6.{name}.has_tail', '1234' in out) + +# T7: 不误伤 Authorization header / Bearer +for msg in ['Authorization header: bearer_user_token_xxxxxxx', 'Bearer raw_token_zzz']: + rec = logging.LogRecord('t', logging.INFO, '', 0, msg, None, None) + f.filter(rec) + ok(f'T7.unmodified_{msg[:15]}', rec.getMessage() == msg, f'out={rec.getMessage()!r}') + +# T8: 端到端 admin PUT roundtrip → client GET 一致 +roundtrip_token = 'rt_secret_RT99' +r = c.put( + '/api/v1/admin/credential-slot/', + data=json.dumps({'app_id': 'roundtrip_test', 'access_token': roundtrip_token}), + content_type='application/json', + HTTP_AUTHORIZATION=f'Bearer {admin_token}', +) +ok('T8.put_200', r.status_code == 200, f'sc={r.status_code}') +# admin GET 应脱敏 +r = c.get('/api/v1/admin/credential-slot/', HTTP_AUTHORIZATION=f'Bearer {admin_token}') +b = r.json() +ok('T8.admin_get_masked', b['data'].get('access_token') != roundtrip_token, + f'admin GET 应脱敏 got={b["data"].get("access_token")!r}') +ok('T8.admin_get_tail_RT99', b['data'].get('access_token', '').endswith('RT99')) +# client GET 应明文 +r = c.get('/api/credential-slot/', HTTP_AUTHORIZATION=f'Bearer {user_token}') +b = r.json() +ok('T8.client_get_plain', b['data'].get('access_token') == roundtrip_token, + f'client GET 应明文 got={b["data"].get("access_token")!r}') +ok('T8.client_app_id', b['data'].get('app_id') == 'roundtrip_test') + +# T9: 端到端 logger.info 真打印 → console 脱敏（防御性兜底真实生效） +log = logging.getLogger('aiapp') +console = next((h for h in log.handlers if 'StreamHandler' in type(h).__name__ and 'Aliyun' not in type(h).__name__), None) +if console is None: + FAIL.append('T9.no_console_handler 找不到 console handler') +else: + buf = io.StringIO() + orig = console.stream + console.stream = buf + try: + log.info('防御性测试：access_token=defensive_secret_DEFC') + finally: + console.stream = orig + out = buf.getvalue() + ok('T9.logger_info_no_plain', 'defensive_secret_' not in out, f'out={out.strip()!r}') + ok('T9.logger_info_tail', 'DEFC' in out) + +# 还原 + 清理 +slot = reset_probe() +ok('T_FINAL.db_restored.app_id', slot.app_id == 'probe_app') +ok('T_FINAL.db_restored.access_token', slot.access_token == 'probe_secret_xxxx') +cache.delete(f'admin_token:{admin_token}') +cache.delete(f'token:{user_token}') + +print('=' * 70) +print(f'PASS ({len(PASS)}):') +for p in PASS: + print(f' ✓ {p}') +if FAIL: + print(f'FAIL ({len(FAIL)}):') + for x in FAIL: + print(f' ✗ {x}') + raise SystemExit(1) +print('ALL PASS') +``` + +**执行 + 落地证据 + 删除**： + +```bash +# 1. 执行验收 +python manage.py shell < _phase3_02_verify.py + +# 2. 把 stdout 复制粘贴进 SUMMARY.md（PASS 列表作为证据） +# 建议用 PowerShell： +# python manage.py shell < _phase3_02_verify.py 2>&1 | Tee-Object -FilePath .planning/phases/03-client-and-log-mask/03-VERIFICATION.md.tmp + +# 3. 验收完毕删除三个临时脚本（_phase3_01_verify.py + _phase3_02_unit_test.py + _phase3_02_verify.py） +rm _phase3_01_verify.py +rm _phase3_02_unit_test.py +rm _phase3_02_verify.py +``` + +**严格禁止做的事**： +- ✗ 不要把 `_phase3_*.py`（含 `_phase3_01_verify.py` / `_phase3_02_unit_test.py` / `_phase3_02_verify.py`）git add（验收一次性产物，不入仓库） +- ✗ 不要把 `roundtrip_token = 'rt_secret_RT99'` / `defensive_secret_DEFC` 等测试值写到 SUMMARY 之外的可被搜索的文档（只在脚本中即时生成、cache 自动 release） +- ✗ 不要在脚本里依赖 daphne / runserver 进程（用 `django.test.Client` in-process） +- ✗ 不要跳过「还原 DB 探针态」步骤 —— 必须在脚本结束前 `app_id='probe_app' / access_token='probe_secret_xxxx'`，给后续工作留稳定起点 +- ✗ 不要忘了 cache.delete 清理临时 token key + + + + +# 1. 跑端到端验收（脚本退出码即结果） +python manage.py shell < _phase3_02_verify.py + +# 2. 验收完毕删除三个临时脚本 +python -c "import os; [os.remove(f) for f in ['_phase3_01_verify.py', '_phase3_02_unit_test.py', '_phase3_02_verify.py'] if os.path.exists(f)]; print('OK: 临时脚本已删除')" + +# 3. 确认 working tree 中无 _phase3_*.py 残留 +python -c "import os, glob; assert not glob.glob('_phase3_*.py'), '仍有残留临时脚本'; print('OK: working tree 清洁')" + +# 4. 确认 DB 探针态保持 +python manage.py shell -c "from aiapp.models import CredentialSlot; s = CredentialSlot.objects.get(pk=1); assert s.app_id == 'probe_app' and s.access_token == 'probe_secret_xxxx', f'DB 探针未还原 {s.app_id}/{s.access_token}'; print('OK: DB probe restored')" + + + + + - `_phase3_02_verify.py` 输出 `ALL PASS`，PASS 行数 ≥ 25（9 truth × 平均 2-3 断言） + - FAIL 行数 = 0 + - DB 探针态：`pk=1 / app_id='probe_app' / access_token='probe_secret_xxxx'` + - Redis 中 `admin_token:*` / `token:*` 测试 key 已 cache.delete + - working tree 中无 `_phase3_*.py` 残留（三个临时脚本均已删除） + - 验收输出已被人工复制粘贴到 SUMMARY.md（task 4 会引用） + + + + 9 条端到端 truth 全 PASS；DB 探针还原；临时 token 清理；三个临时验收脚本（`_phase3_01_verify.py` / `_phase3_02_unit_test.py` / `_phase3_02_verify.py`）已删除。 + + + + + Task 4: 顶部追加 docs/修改记录.md 条目（覆盖 client view + filter + LOGGING 三处改动） + docs/修改记录.md + + + 1. **完整读 `docs/修改记录.md` 行 1-80**（已读）—— 确认头部「修改格式说明」段 + 现有 3 条 Phase 1/2 条目；新条目追加在「修改历史」段最顶部，紧邻 `[2026-05-07] Phase 2` 之上 + 2. 读 CLAUDE.md「项目修改记录规则」+「`qy_lty` 与 `qy-lty-admin` 是独立项目，各自维护」段 —— 确认本 phase 不写 qy-lty-admin 互引（CONTEXT 锁定决策 + RESEARCH 实证） + 3. 读 RESEARCH「附录：access_token 实际泄露路径详细分析」—— 把「防御性兜底」语义写入修改原因 + 4. 读 `.planning/phases/03-client-and-log-mask/03-01-SUMMARY.md`（plan 03-01 写的）—— 确认 client view 的实际行号 / 文件 + 5. **执行前**记录 phase 启动时点：把 `qy-lty-admin/docs/修改记录.md` 的 `os.path.getmtime(...)` 写入临时变量（用于本 task verify #4 的 mtime 比较，替代 git diff cwd='..'）。命令：`python -c "import os, json, pathlib; mt = os.path.getmtime('../qy-lty-admin/docs/修改记录.md') if os.path.exists('../qy-lty-admin/docs/修改记录.md') else None; pathlib.Path('_phase3_admin_mtime.txt').write_text(str(mt)); print(mt)"`（**在改 docs/修改记录.md 之前**跑这一行；mtime 落在 `_phase3_admin_mtime.txt`，验收完一并删） + + + + +在 `docs/修改记录.md` 行 25 附近（紧邻 `### [2026-05-07] Phase 2 — 管理端通用凭据槽位 REST 接口（GET 脱敏 / PUT 覆写）` **之上**、`` **之下**）插入以下条目（保持「最新在最前」顺序）： + +```markdown +### [2026-05-08] Phase 3 — 客户端凭据槽位 GET 接口 + 阿里云日志 access_token 脱敏 + +配套 Phase：[.planning/phases/03-client-and-log-mask/](.planning/phases/03-client-and-log-mask/) +覆盖需求：CRED-05 + CRED-06 +设计参考：1:1 复刻 `aiapp.views.CredentialSlotAdminView` 的 GET 部分（删 `_ensure_admin` / `_build_response_data` / PUT 三处），实现明文返回客户端 view；新建 `common/logging/filters.py:AccessTokenMaskFilter` 作为 LOGGING.handlers 层防御性兜底 + +- **文件路径**: + - `aiapp/views.py`（修改 — 文件末尾追加 `_credential_slot_client_data_schema` 客户端响应 schema + `CredentialSlotClientView` APIView 类，仅 GET，明文返回；imports 段未动；Phase 2 既有 `CredentialSlotAdminView` 未动） + - `qy_lty/urls.py`（修改 — imports 段追加 `from aiapp.views import CredentialSlotClientView`；`api_urlpatterns` 列表中追加 `path('credential-slot/', CredentialSlotClientView.as_view(), name='client_credential_slot')`，注册位置：`common/upload/` 之后、`v1/admin/` 之前） + - `common/logging/__init__.py`（**新建** — 空文件，让 `common.logging` 成为可 import 的 Python 包） + - `common/logging/filters.py`（**新建** — `AccessTokenMaskFilter(logging.Filter)` 类 + 4 个 regex 模式（JSON / Python dict repr / URL query / 等号或冒号兜底）+ `filter()` 方法重写 `record.msg` 与 `record.args` 中的 access_token 字段值为 `mask_token(value)` 输出） + - `qy_lty/settings.py`（修改 — `LOGGING` 字典新增 `'filters'` 段（用 `'()': 'common.logging.filters.AccessTokenMaskFilter'` dictConfig 工厂语法）；`'handlers'.aliyun` 与 `'handlers'.console` 各追加 `'filters': ['access_token_mask']`；loggers 段 5 条 logger 完全未动） +- **修改类型**: 新增 +- **修改内容**: + - 暴露 `GET /api/credential-slot/`（路径与管理端 `/api/v1/admin/credential-slot/` **完全分开**，客户端走 `/api/` 一级命名空间不进 `v1/admin/` 子路径）：`RedisTokenAuthentication` + `IsAuthenticated`，**不**做 is_staff 二次校验（admin / user token 都允许；admin 用户是手机用户超集，CONTEXT 锁定决策）；返回 `{ success, code, message, data: { app_id, access_token: <**明文**>, updated_at } }`，Access Token 直接返回 `serializer.data`（不调 `mask_token`），供手机端（LTY_App_Project_URP）/ 设备端（LTY_Project）实际调用阿里云 / 火山 / 腾讯第三方服务 + - 新建 `AccessTokenMaskFilter`：4 个正则模式覆盖 JSON 字符串（`"access_token":"VALUE"`）、Python dict repr（`'access_token':'VALUE'`）、URL query（`access_token=VALUE`）、等号或冒号兜底（`access_token: VALUE`）共 4 种序列化形态；filter 同时改 `record.msg` 与 `record.args`（避免 Formatter 阶段再用 `%` 拼接出明文，per RESEARCH Pitfall 2）；只匹配 `access_token` 字段名为前缀锚点，**不**误伤 `Authorization header:` / `Bearer` / 裸 user token（per RESEARCH Pitfall 3）；filter 永远 `return True` 不丢弃 record（per RESEARCH Pitfall 1） + - LOGGING dictConfig 注册：filter 段用 `'()': '...'` 工厂语法（不是 `'class'`，per RESEARCH Pitfall 5）；filter 挂在 `handlers.aliyun` / `handlers.console` 两个 handler 上（**不**挂 loggers 段，per RESEARCH Pitfall 1 — 挂 logger 仅过滤直接通过该 logger 的 record，挂 handler 才统一覆盖所有 logger → handler 路径）；既有 5 条 logger 配置完全未动 + - Swagger / ReDoc 自动暴露：method-level `@swagger_auto_schema` 装饰器；响应 data schema 用独立 `_credential_slot_client_data_schema`，access_token 字段 description 显式标注「明文 Access Token，供手机/设备端实际调用第三方服务（管理端同接口会脱敏返回末 4 位）」，避免前端误解明文 / 脱敏 + - 不引入新依赖（沿用 Django 4.2.13 + DRF + drf-yasg + Phase 1/2 落地的 `CredentialSlot.get_solo` / `CredentialSlotSerializer` / `mask_token`） +- **修改原因**: Milestone v1.0「通用凭据槽位（APP ID + Access Token）」Phase 3 收尾 phase — 同时落地客户端读取（CRED-05）与日志脱敏（CRED-06）。客户端读取需要明文（手机/设备端 Unity 调阿里云 / 火山 / 腾讯 SDK 时第三方 API 校验 token 字符级一致），所以 view 层不脱敏；但「明文走 view」会让任何后续开发者写 `logger.info(f"PUT body: {request.data}")` 类代码立即把 access_token 打到阿里云日志服务，所以新增 LOGGING.handlers 层 filter 作为防御性兜底。RESEARCH 已实证：当前仓库**没有**任何代码 logger 输出 `CredentialSlot.access_token` 明文（`StandardResponseMiddleware` 不打日志、view 不显式 logger 字段、Django 默认 access log 不含 body），所以 CRED-06 的端到端验证靠**单元测试**伪造 LogRecord 验证 filter 行为（4 种序列化形态 + 不误伤 Authorization 字段）+ 1 条端到端 logger.info 真实输出脱敏验证，不靠端到端找泄露路径。这是 CRED-06 的真实价值 — 防御性兜底，让未来代码改动天然安全 +- **跨项目联动**: 无 — 客户端 GET `/api/credential-slot/` 给 Unity 客户端（`LTY_Project` / `LTY_App_Project_URP`）使用，那两个 repo 各自维护修改记录，不在本仓库范畴；`qy-lty-admin`（Web 管理后台前端）**不消费**此接口（管理端走 Phase 2 落地的 `/api/v1/admin/credential-slot/`，由 admin token 鉴权 + 脱敏返回）。CLAUDE.md 跨项目规则下：本 phase 既不影响 qy-lty-admin 也不与 Unity 客户端在同一仓库，故不在 qy-lty-admin/docs/修改记录.md 写互引条目；Unity 客户端改动由 LTY_Project / LTY_App_Project_URP 在自身仓库各自记录 +- **后续动作**: Milestone v1.0 至此完成；下一周期 milestone 候选见 `.planning/REQUIREMENTS.md` 「候选优先级」段（HIGH：ACH-02 / SMS 频率限制 / DEBUG 收紧 / 测试基础设施 / 测试 MAC 硬编码；MEDIUM：好感度 P2-P4 / Python 版本升级 / device_interaction 拆分） + +``` + +**严格禁止做的事**： +- ✗ 不要在 `qy-lty-admin/docs/修改记录.md` 写互引条目（CONTEXT 锁定 + RESEARCH 实证；本 phase 与 qy-lty-admin 无 API 联动） +- ✗ 不要写「跨项目联动: 待补充」（明确写「无 —— 客户端给 Unity 用，不在 qy-lty-admin 范畴」） +- ✗ 不要把 4 处文件改动拆成 4 条独立条目（一个 phase = 一条修改记录条目，覆盖所有改动文件） +- ✗ 不要修改既有 Phase 1 / Phase 2 的 3 条条目（仅追加新条目） +- ✗ 不要把日期写成 `[2026-05-07]`（今天是 2026-05-08） +- ✗ 不要把临时脚本 `_phase3_*.py` 写进修改记录（不属于代码改动，已删除） + + + + +# 1. 顶部已追加 Phase 3 条目 +python -c "src=open('docs/修改记录.md',encoding='utf-8').read(); idx_p3 = src.find('[2026-05-08] Phase 3'); idx_p2 = src.find('[2026-05-07] Phase 2'); assert 0 < idx_p3 < idx_p2, f'Phase 3 条目位置错误：p3={idx_p3} p2={idx_p2}'; print('OK: Phase 3 条目在 Phase 2 之上（最新在最前）')" + +# 2. 5 处文件改动都被记录 +python -c "src=open('docs/修改记录.md',encoding='utf-8').read(); ph3 = src[src.find('[2026-05-08] Phase 3'):src.find('[2026-05-07] Phase 2')]; required = ['aiapp/views.py', 'qy_lty/urls.py', 'common/logging/__init__.py', 'common/logging/filters.py', 'qy_lty/settings.py']; missing = [f for f in required if f not in ph3]; assert not missing, f'缺文件: {missing}'; print('OK: 5 处文件改动均已记录')" + +# 3. 跨项目联动字段存在且明确写「无」 +python -c "src=open('docs/修改记录.md',encoding='utf-8').read(); ph3 = src[src.find('[2026-05-08] Phase 3'):src.find('[2026-05-07] Phase 2')]; assert '**跨项目联动**' in ph3, '缺跨项目联动字段'; assert ('无 ' in ph3 or '无—' in ph3 or '无—' in ph3), '跨项目联动应明确写「无」'; print('OK: 跨项目联动字段已明确')" + +# 4. 不写互引：用 mtime 比对替代 git diff（qy_lty 与 qy-lty-admin 是独立 repo，git diff cwd='..' 在父目录非 repo 时会 fatal 被 || 兜掉等于空检查）。 +# 比较 qy-lty-admin/docs/修改记录.md 的 mtime 与 phase 启动时记录的 mtime；若一致则未被本 phase 改动。 +# 若 _phase3_admin_mtime.txt 缺失（read_first 漏跑），降级为「acceptance_criteria 人工核对」（非 BLOCK）。 +python -c " +import os, pathlib +admin_md = '../qy-lty-admin/docs/修改记录.md' +mtime_file = '_phase3_admin_mtime.txt' +if not os.path.exists(admin_md): + print('OK: qy-lty-admin/docs/修改记录.md 不存在（独立 repo 未克隆）—— 自然不可能被改') +elif not os.path.exists(mtime_file): + print('SKIP: _phase3_admin_mtime.txt 缺失，降级为人工核对（见 acceptance_criteria）') +else: + expected = pathlib.Path(mtime_file).read_text().strip() + actual = str(os.path.getmtime(admin_md)) + assert expected == actual, f'qy-lty-admin/docs/修改记录.md 在本 phase 中被改动（启动 mtime={expected} 当前 mtime={actual}）' + print(f'OK: qy-lty-admin/docs/修改记录.md mtime 未变（{actual}）') +" + +# 5. CRED-05 + CRED-06 都被显式提及 +python -c "src=open('docs/修改记录.md',encoding='utf-8').read(); ph3 = src[src.find('[2026-05-08] Phase 3'):src.find('[2026-05-07] Phase 2')]; assert 'CRED-05' in ph3 and 'CRED-06' in ph3; print('OK: CRED-05 + CRED-06 均显式标注')" + +# 6. 清理 phase 启动时记录的 mtime 临时文件 +python -c "import os; [os.remove(f) for f in ['_phase3_admin_mtime.txt'] if os.path.exists(f)]; print('OK: _phase3_admin_mtime.txt 已删除')" + + + + + - `docs/修改记录.md` 顶部含 `### [2026-05-08] Phase 3 — 客户端凭据槽位 GET 接口 + 阿里云日志 access_token 脱敏` 条目 + - 条目位置：在 `` 之下、`### [2026-05-07] Phase 2 ...` 之上 + - 5 处文件路径全部列出（`aiapp/views.py` / `qy_lty/urls.py` / `common/logging/__init__.py` / `common/logging/filters.py` / `qy_lty/settings.py`） + - 修改类型 / 修改内容 / 修改原因 / 跨项目联动 4 段齐全（按 CLAUDE.md 规定格式） + - 跨项目联动字段明确写「无 —— 客户端给 Unity (LTY_Project / LTY_App_Project_URP) 用」 + - 显式提及 CRED-05 + CRED-06 两个需求 ID + - **人工核对**：在另一个终端 `cd ..\qy-lty-admin && git status` 应显示 `docs/修改记录.md` 未在 unstaged / staged 列表中（qy-lty-admin 与 qy_lty 是独立 repo，本 phase 不应触碰对方 docs；mtime 自动检查作为兜底但人工核对是权威） + - 既有 Phase 1 / Phase 2 三条条目原文未动 + - phase 启动时点的 `_phase3_admin_mtime.txt` 已删除 + + + + Phase 3 修改记录条目已追加到 `docs/修改记录.md` 顶部；5 处文件改动 + CRED-05/06 + 跨项目联动「无」均已记录；qy-lty-admin/docs/修改记录.md 的 mtime 未变（自动 + 人工双重验收）；`_phase3_admin_mtime.txt` 临时文件清理。 + + + + + + +## Trust Boundaries + +| Boundary | Description | +|----------|-------------| +| logger 调用方 → handler | 任何 view / middleware / 第三方库可能 `logger.info(包含明文 access_token 的字符串)`；filter 在 handler 入口拦截脱敏 | +| handler → 阿里云日志服务 | `AliyunLogHandler.emit` 调 `record.getMessage()`，filter 已重写 record.msg 后 emit 自然拿脱敏值 | +| handler → stderr | `console` handler 的 stream 输出，filter 同样兜底 | + +## STRIDE Threat Register + +| Threat ID | Category | Component | Disposition | Mitigation Plan | +|-----------|----------|-----------|-------------|-----------------| +| T-03-08 | Information Disclosure | 未来开发者 `logger.info(f"PUT body: {request.data}")` 把 access_token 明文打到阿里云日志 | mitigate | `AccessTokenMaskFilter` 在 LOGGING.handlers.aliyun 兜底；4 种序列化形态全覆盖（JSON / Pyrepr / Query / Fallback）| +| T-03-09 | Information Disclosure | 误把 Authorization Bearer token / 裸 user-token 当 access_token 脱敏 | mitigate | filter 4 个 regex 全部以 `access_token` 字段名为前缀锚点；测试 T7 显式验证 `Authorization header:` / `Bearer` 字段未被改写 | +| T-03-10 | Tampering | filter 写错丢弃 record | mitigate | filter() 永远 return True；单元测试验证 | +| T-03-11 | DoS | 正则匹配在超长字符串上耗 CPU | accept | 4 个正则均用限定终止符（`[^"]+` / `[^']+` / `[^&\s"\']+`），不用贪婪 `.+`；`_NEEDLE` 短路在大多数无 `access_token` 的 record 上 0 成本 | +| T-03-12 | Repudiation | filter 改写后无法回溯原始 record | accept | 改写仅针对 access_token 字段值，原始 logger 调用栈、级别、文件名、行号全部保留；阿里云日志的 levelno / pathname / filename / funcName / lineno 字段不动 | +| T-03-13 | Elevation of Privilege | filter 类被恶意修改 | accept | `common/logging/filters.py` 在仓库内，受 git + code review 保护；与其它仓库代码同等信任级别 | +| T-03-14 | Spoofing | dictConfig 工厂语法写错让 filter 不加载 | mitigate | Task 2 验收 step 2-3 显式断言 `Django setup() 不报 ValueError` + filter 实例真实挂载到 handler.filters 列表 | + + + +**Plan 整体验收（汇总 task 1/2/3/4 的自动化检查）**： + +```bash +# Task 1: 包 + filter 类 + 单元测试脚本 +python -c "from common.logging.filters import AccessTokenMaskFilter; assert AccessTokenMaskFilter.__bases__[0].__name__ == 'Filter'; assert len(AccessTokenMaskFilter._PATTERNS) == 4; print('OK: filter 类 + 4 模式')" +python _phase3_02_unit_test.py # task 3 后此文件已删，仅 task 1 执行后可跑 + +# Task 2: settings.py + Django 启动 +python -c "src=open('qy_lty/settings.py',encoding='utf-8').read(); assert 'access_token_mask' in src; assert src.count(\"['access_token_mask']\") >= 2; print('OK: settings filters 段 + 2 handler 引用')" +python -c "import django, os; os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'qy_lty.settings'); django.setup(); print('OK: dictConfig 加载无 ValueError')" + +# Task 3: 端到端 9 truth + 临时脚本删除 +python manage.py shell < _phase3_02_verify.py # 此前 task 3 已跑过；此处复跑前需先重写脚本 +ls _phase3_*.py 2>/dev/null && echo "残留脚本！" || echo "OK: 无残留" + +# Task 4: 修改记录条目 +python -c "src=open('docs/修改记录.md',encoding='utf-8').read(); assert '[2026-05-08] Phase 3' in src; assert 'CRED-05' in src and 'CRED-06' in src; print('OK: 修改记录条目')" +``` + + + +- [ ] `common/logging/__init__.py`（空文件）+ `common/logging/filters.py`（含 `AccessTokenMaskFilter` 类 + 4 regex + `filter()` 方法）已创建 +- [ ] `_phase3_02_unit_test.py` 输出 `ALL UNIT TESTS PASS`，覆盖 4 形态脱敏 + 不误伤 Authorization / Bearer + tuple args 脱敏 + 短 token 全脱 + filter() return True +- [ ] `qy_lty/settings.py:LOGGING` 字典含 `'filters'` 段用 `'()': 'common.logging.filters.AccessTokenMaskFilter'` 工厂语法 +- [ ] `LOGGING.handlers.aliyun` / `LOGGING.handlers.console` 各引用 `filters: ['access_token_mask']`，loggers 段 5 条 logger 完全未动 +- [ ] `django.setup()` 不报 `ValueError: Unable to configure filter ...` +- [ ] 端到端：`logger.info('access_token=secret_xxxx_ABCD')` 后 console 输出含 `ABCD` 不含 `secret_xxxx_` +- [ ] `_phase3_02_verify.py` 输出 `ALL PASS`，9 条 truth（含 plan 03-01 的 5 条 client view + plan 03-02 的 3 条 filter 单元 + 1 条端到端 logger 真打印）全过 +- [ ] `_phase3_01_verify.py` / `_phase3_02_unit_test.py` / `_phase3_02_verify.py` / `_phase3_admin_mtime.txt` 临时文件均已删除（git status 干净） +- [ ] DB 探针态 `pk=1 / app_id='probe_app' / access_token='probe_secret_xxxx'` 保持 +- [ ] `docs/修改记录.md` 顶部含 `[2026-05-08] Phase 3` 条目；5 处文件 + CRED-05/06 + 跨项目联动「无」全标注 +- [ ] `qy-lty-admin/docs/修改记录.md` 未被本 phase 改动（mtime 自动 + 人工 `cd ..\qy-lty-admin && git status` 双重验收，CONTEXT 锁定 + RESEARCH 实证） +- [ ] Phase 1 / Phase 2 既有代码与既有修改记录条目原文未动 + + + +完成后创建 `.planning/phases/03-client-and-log-mask/03-02-SUMMARY.md`，记录： +- 改动 5 处文件的实际行号区间（含 `__init__.py` 空文件大小、`filters.py` 总行数、settings.py LOGGING 块前后行差） +- `_phase3_02_verify.py` 输出（PASS 列表完整粘贴；FAIL 必须为 0） +- 防御性兜底语义说明（RESEARCH 已实证当前仓库无 access_token 泄露路径，filter 是为未来代码改动留的安全网） +- Milestone v1.0 完结声明：CRED-01 至 CRED-06 全部 Done，REQUIREMENTS.md 同期更新 +- 同步更新 `.planning/STATE.md` 与 `.planning/ROADMAP.md`：Phase 3 标记 ✓ 完成，progress 100%，next 步骤为下一周期 milestone 候选评估 + +