52 KiB
Raw Blame History

Phase 3客户端读取与日志脱敏 - Research

Researched: 2026-05-08 Domain: Django 4.2 客户端 REST 接口 + Python logging.Filter 日志脱敏 Confidence: HIGH所有结论均通过 read 实证仓库源码得出)

Summary

本 phase 在仓库现状下属于纯结构落地 —— Phase 1 / Phase 2 已经把模型、序列化器、mask_tokenget_solo()success_responseRedisTokenAuthenticationget_standardized_response_schema 等所有依赖件全部备齐。CRED-05 的实现是复刻 CredentialSlotAdminView,删 PUT、删 _ensure_admin、删 _build_response_data 中的 mask_token 调用这三步即可。

CRED-06日志脱敏的研究核心结论当前生产中实际会泄露 access_token 明文的代码路径只有 1 条(不是 CONTEXT 假设的 3 条),即 userapp/authentication.py:23logger.debug(f"Authorization header: {token}"),且该路径上的 token 是 user/admin tokenAuthorization 头),不是 CredentialSlot.access_token 字段。CredentialSlot.access_token 没有任何代码路径会经 logger 输出 —— StandardResponseMiddleware 不打日志见验证view 层只 logger.warning 用户 id不打 access_token 字段Django 默认 access log 不包含 body。

但 CRED-06 的存在意义仍然成立:为未来防御性兜底。任何后续开发者写 logger.info(f"PUT body: {request.data}") 这类语句,都会一次性把 access_token 明文打到 Aliyun 日志服务INFO 级 → aliyun handlerAccessTokenMaskFilter 挂在 LOGGING.handlers 上是这层防御的标准实现。

Primary recommendationCRED-05 在 aiapp/urls.py 加一行;客户端命名空间根据现有约定走 qy_lty/urls.py:50path('ai/', include('aiapp.urls')) 已经收容的子 urls —— 但 CONTEXT 锁定路径是 /api/credential-slot/(不带 ai/ 前缀),所以实际选定的注册位置是直接在 qy_lty/urls.pyapi_urlpatterns 列表中加一行 path('credential-slot/', CredentialSlotClientView.as_view(), ...)详见下文「客户端路由汇总点」。CRED-06 在 common/logging/filters.py新建文件,含 __init__.py)里实现 AccessTokenMaskFilter

Architectural Responsibility Map

Capability Primary Tier Secondary Tier Rationale
客户端 GET 凭据明文 API / BackendDRF APIView 与 Phase 2 同位置aiapp/views.py同 model 同 serializer行为反向
鉴权识别 user token 横切(已有 RedisTokenAuthentication 复用 Phase 2 已 import
路由注册 项目级 url 顶层qy_lty/urls.py CONTEXT 锁定 /api/credential-slot/ 不在任何 app 的 sub-namespace 下
日志脱敏 filter 横切common/logging/ LOGGING handler 配置settings.py 单一 filter 类挂到所有 handler覆盖所有 logger 路径
修改记录条目 docs/修改记录.md仅 qy_lty CONTEXT 明确不写 qy-lty-admin 互引Unity 客户端不在 qy-lty-admin

<user_constraints>

User Constraints (from CONTEXT.md)

Locked Decisions

CRED-05 客户端 GET 接口:

  • 路径:/api/credential-slot/(与管理端 /api/v1/admin/credential-slot/ 完全分开;客户端走 /api/ 一级命名空间)
  • HTTP 方法:仅 GET客户端不能写
  • View 类:CredentialSlotClientView(与 CredentialSlotAdminView 形成对称命名)
  • View 实现:自定义 APIView + 仅 get 方法
  • View 放置位置:aiapp/views.py 末尾(与 CredentialSlotAdminView 紧邻)
  • 鉴权:authentication_classes = [RedisTokenAuthentication] + permission_classes = [IsAuthenticated]
  • 不做 is_staff 校验 —— 手机/设备 user token 与 admin token 都允许
  • 响应序列化器:直接复用 Phase 2 的 CredentialSlotSerializer(不新建客户端专用)
  • 客户端 view 行为:success_response(data=serializer.data) 直返明文(不调 mask_token
  • Swaggermethod-level @swagger_auto_schema + get_standardized_response_schemaresponse description 标注「明文 APP ID + Access Token」

CRED-06 日志脱敏:

  • 走「路径 A」自定义 logging.Filter 实现
  • 文件位置:common/logging/filters.py(新建)
  • 类名:AccessTokenMaskFilter(logging.Filter)
  • 注册位置:settings.LOGGING.filters + 挂到 handlers.aliyun / handlers.console
  • 不动 StandardResponseMiddleware
  • 正则建议覆盖 4 种序列化形态JSON 字符串 / Python dict repr / query string / 其他

Claude's Discretion

  • filter 用 logging.Filter 子类还是 logging.Formatter 子类(前者改 record后者改格式化输出 —— 推荐 Filter 更易测)
  • 是否在 StandardResponseMiddleware 加辅助保险(视 researcher 看到 middleware 实际是否输出日志而定 —— 本 RESEARCH 已证实 middleware 不打日志,故无需加

Deferred Ideas (OUT OF SCOPE)

  • DB at-rest 加密 access_token 字段(留 v2.x
  • 客户端调用频率限流
  • token 轮换 / refresh token 机制
  • Unity 客户端 SDK 联调(在独立 repo
  • 生产日志脱敏自动化测试CI
  • DEBUG / CORS_ALLOW_ALL_ORIGINS 收紧(独立 phase </user_constraints>

<phase_requirements>

Phase Requirements

ID Description Research Support
CRED-05 客户端 GET /api/credential-slot/user token 鉴权(token:{token} Redis key 体系,复用 RedisTokenAuthentication);明文返回 { app_id, access_token, updated_at } aiapp/views.py:600-687 提供 CredentialSlotAdminView 完整 1:1 模板(删 PUT、删 _ensure_admin、删 mask_token 调用即可);get_user_id_from_token 已支持 admin/user token 双查(userapp/utils.py:39-45),双 token 持有者都能通过 IsAuthenticated
CRED-06 Access Token 日志过滤:阿里云日志格式化器 / 自定义日志过滤器中识别 access_token 字段并脱敏,覆盖 PUT 请求体、admin GET 响应体两条最易泄露路径 LOGGING 配置位于 qy_lty/settings.py:372-412;阿里云 handler 类:common.aliyun_logging.AliyunLogHandlersettings.py:378mask_token 已存在于 common/utils.py:10;当前仓库零处自定义 logging.Filter,需新建 common/logging/filters.py(含 __init__.py
</phase_requirements>

Standard Stack

Core已就位无需 install

Library Version Purpose Why Standard
Django 4.2.13 项目主框架 [VERIFIED: settings.py:4 注释 + qy_lty/.planning/codebase/STACK.md]
DRF (rest_framework) 3.xunpinned REST 视图层 [VERIFIED: requirements.txt + INSTALLED_APPS]
drf-yasg unpinned Swagger schema 自动生成 [VERIFIED: settings.py:529-554 + qy_lty/urls.py:22-24]
aliyun-log-python-sdk unpinned 阿里云日志 SDK提供 LogClient / PutLogsRequest / LogItem [VERIFIED: common/aliyun_logging.py:2 + requirements.txt]
Python logging stdlib filter / formatter / handler 基础设施 [VERIFIED: stdlibDjango LOGGING dictConfig 直接调用]

Supporting已就位

Library Version Purpose When to Use
common.utils.mask_token 项目内 末 4 位脱敏 filter 内调用,复用 Phase 1 实现
common.responses.success_response 项目内 标准 200 壳层 view return
common.swagger_utils.get_standardized_response_schema 项目内 drf-yasg 标准响应 schema swagger_auto_schema responses
userapp.authentication.RedisTokenAuthentication 项目内 Bearer token 双查admin / user view authentication_classes

Alternatives Considered不采用

Instead of Could Use Tradeoff
logging.Filter 子类 logging.Formatter 子类 Formatter 改格式化输出文本,对 record.args 是 dict / tuple 时改不动结构化字段Filter 改 record.msg + record.args 早于 Formatter更彻底CONTEXT 已锁定 Filter此条仅备查
独立 libpython-json-logger 引入新依赖,违反 CONTEXT 「不引入新依赖」
在 view / middleware 内显式脱敏 覆盖面窄、侵入大CONTEXT 已否决

版本验证:本 phase 不安装新包,无需跑 npm view / pip show。所有依赖在 requirements.txt 中已列出unpinned生产已运行的版本即为目标版本。

Architecture Patterns

System Architecture Diagram

┌─────────────────────────────────────────────────────────────────────────┐
│  [客户端] Unity Mobile/Device                                            │
│           │                                                              │
│           │ GET /api/credential-slot/   Authorization: Bearer <token>    │
│           ▼                                                              │
│  ┌──────────────────────────────────────────────────────────────────┐   │
│  │ Django MIDDLEWARE chain (settings.py:82-95)                       │   │
│  │  SecurityMiddleware → CorsMiddleware → AuthenticationMiddleware   │   │
│  │  → StandardResponseMiddleware (不打日志)                          │   │
│  └──────────────────────────────────────────────────────────────────┘   │
│           │                                                              │
│           ▼                                                              │
│  qy_lty/urls.py:66  path('api/', include(api_urlpatterns))              │
│           │                                                              │
│           ▼                                                              │
│  qy_lty/urls.py:NEW  path('credential-slot/', CredentialSlotClientView)│
│           │                                                              │
│           ▼                                                              │
│  ┌──────────────────────────────────────────────────────────────────┐   │
│  │ CredentialSlotClientView (aiapp/views.py末尾新增)               │   │
│  │  authentication_classes = [RedisTokenAuthentication]              │   │
│  │   └─ 双查 admin_token:{token} → token:{token} (utils.py:39-45)   │   │
│  │      logger.debug(f"Authorization header: {token}")  ⚠ 已存在泄露 │   │
│  │  permission_classes = [IsAuthenticated]   (admin & user 都通过)  │   │
│  │  get():                                                           │   │
│  │    instance = CredentialSlot.get_solo()                           │   │
│  │    serializer = CredentialSlotSerializer(instance)                │   │
│  │    return success_response(data=serializer.data)  ← 明文 access_token│
│  └──────────────────────────────────────────────────────────────────┘   │
│                                                                          │
│  并行Python logging chain (settings.py:372-412)                        │
│   logger ('aiapp' / 'userapp' / 'common' / 'django')                    │
│      │                                                                   │
│      ▼ record                                                            │
│   handlers: ['aliyun', 'console']                                        │
│      │      ↑ 新增 filters: ['access_token_mask']                       │
│      │                                                                   │
│      ▼                                                                   │
│   AccessTokenMaskFilter.filter(record):  ← 新建common/logging/filters.py│
│      重写 record.msg + record.args 中的 access_token 值                  │
│      ▼                                                                   │
│   AliyunLogHandler.emit(record) → 阿里云日志服务(脱敏后)              │
│   StreamHandler.emit(record) → stderr脱敏后                         │
└─────────────────────────────────────────────────────────────────────────┘
qy_lty/
├── qy_lty/
│   ├── urls.py            # [修改] api_urlpatterns 加 credential-slot/ 路径
│   └── settings.py        # [修改] LOGGING.filters 注册 + handlers.X.filters 引用
│
├── aiapp/
│   └── views.py           # [修改] 末尾追加 CredentialSlotClientView 类
│
├── common/
│   └── logging/           # [新建] 包目录
│       ├── __init__.py    # [新建] 空文件
│       └── filters.py     # [新建] AccessTokenMaskFilter
│
└── docs/
    └── 修改记录.md         # [修改] 顶部追加 Phase 3 条目(仅 qy_lty 端,不写互引)

Pattern 1客户端 view 1:1 模板

模板源aiapp/views.py:600-687CredentialSlotAdminView

# Source: aiapp/views.py:600-656 (Phase 2 已交付,本 phase 1:1 复刻 GET 部分)

class CredentialSlotAdminView(APIView):
    authentication_classes = [RedisTokenAuthentication]   # ← 直接复用
    permission_classes = [IsAuthenticated]                # ← 直接复用
    tags = ['通用凭据槽位(管理端)']

    def _ensure_admin(self, request):                     # ✗ 客户端 view 删
        if not request.user.is_staff:
            return error_response(...)
        return None

    def _build_response_data(self, instance):             # ✗ 客户端 view 删(明文直返)
        serializer = CredentialSlotSerializer(instance)
        data = dict(serializer.data)
        data['access_token'] = mask_token(instance.access_token)  # ← 客户端 view 不脱敏
        return data

    @swagger_auto_schema(...)
    def get(self, request):
        forbidden = self._ensure_admin(request)            # ✗ 客户端 view 删
        if forbidden:
            return forbidden
        instance = CredentialSlot.get_solo()               # ✓ 保留
        data = self._build_response_data(instance)         # → 改为 CredentialSlotSerializer(instance).data
        return success_response(data=data, message="读取成功")  # ✓ 保留

1:1 复刻产物(客户端 view 骨架,仅供 plan 参考;正式行号 / 缩进由 plan 确定):

# 在 aiapp/views.py 末尾追加(紧邻 CredentialSlotAdminView 之后)

class CredentialSlotClientView(APIView):
    """通用凭据槽位客户端读取接口user / admin token 鉴权,明文返回)。

    GET: 返回明文 app_id + access_token供手机/设备端实际调用第三方服务。
    """
    authentication_classes = [RedisTokenAuthentication]
    permission_classes = [IsAuthenticated]
    tags = ['通用凭据槽位(客户端)']

    @swagger_auto_schema(
        operation_description="读取通用凭据槽位(明文 access_token 返回user/admin token 均允许)",
        responses={
            200: openapi.Response('读取成功', get_standardized_response_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="读取成功")

与 admin view 的差异点plan 必须明确实现)

  1. _ensure_admin 整个方法
  2. _build_response_data 整个方法(直接 serializer.data 即可)
  3. 删 PUT method 与 PUT 的 swagger_auto_schema
  4. swagger response data schema 不再_credential_slot_data_schema(脱敏掩码语义);新建一份明文 schemaaccess_token description 标注「明文 Access Token供手机/设备端实际调用第三方服务」

Pattern 2客户端路由注册

当前 qy_lty/urls.py:49-60api_urlpatterns(实测):

# qy_lty/urls.py:49-60 (verbatim)
api_urlpatterns = [
    path('user/', include('userapp.urls')),               # /api/user/*
    path('ai/', include('aiapp.urls')),                   # /api/ai/*
    # path('ali/vi/api/', include('ali_vi_app.urls')),    # 已禁用
    path('device/', include('device_interaction.urls')),  # /api/device/*
    path('card/', include('card.urls')),                  # /api/card/*
    path('achievement/', include('achievement_app.urls')),# /api/achievement/*
    path('food/', include('food_app.urls')),              # /api/food/*
    path('common/upload/', upload_file, name='file-upload'),
    path('v1/admin/', include('userapp.admin_urls')),     # /api/v1/admin/*
]

关键观察

  • 所有客户端命名空间都是 path('<app>/', include('<app>.urls')) 风格user/ai/device/card/achievement/food
  • 没有任何现成的 sub-urls 收容 /api/credential-slot/ —— aiapp/urls.py:8-13 的 urlpatterns 全部是 chat/<bot_id>/ / multichat/ / rtc-chat-history/ / router.urls根本不挂在 /api/<某 app>/credential-slot/ 下,必须放 /api/credential-slot/CONTEXT 锁定)
  • 候选 Aqy_lty/urls.py:api_urlpatterns 直接加):✓ 正确,与 path('common/upload/', upload_file, ...) 同风格 —— 单一 path() 项作为零层后兜底
  • 候选 Baiapp/urls.py 加):✗ 错误,会变成 /api/ai/credential-slot/ 而不是 /api/credential-slot/
  • 候选 C参考现有客户端接口参考 qy_lty/urls.py:57path('common/upload/', upload_file, name='file-upload') —— 单一 view 直接挂到 api_urlpatterns 顶层,无 sub-include这就是本 phase 的 1:1 风格参考

最终选定的 urls.py 文件路径qy_lty/urls.py

最终选定的注册代码行(在 api_urlpatterns 列表内追加,建议位置:紧邻 path('common/upload/', upload_file, name='file-upload'), 之后、path('v1/admin/', include('userapp.admin_urls')), 之前,以保持「客户端散点 → admin 命名空间」的视觉分组):

# qy_lty/urls.py 顶部 imports 段追加:
from aiapp.views import CredentialSlotClientView

# qy_lty/urls.py:49-60 api_urlpatterns 列表中追加:
path('credential-slot/', CredentialSlotClientView.as_view(), name='client_credential_slot'),

Pattern 3AccessTokenMaskFilter 注册骨架

Django 4.2 LOGGING dictConfig 标准结构(含 filters—— [CITED: docs.djangoproject.com/en/4.2/topics/logging/]

LOGGING = {
    'version': 1,
    'disable_existing_loggers': False,

    # 新增段(当前 settings.py:372-412 没有 filters 段,需新增)
    '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': {
        # 现有 4 条 logger 配置无需改动 —— filter 是挂在 handler 上的,不是 logger 上
        '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 },
    },
}

AccessTokenMaskFilter 类骨架common/logging/filters.py

import logging
import re
from common.utils import mask_token


class AccessTokenMaskFilter(logging.Filter):
    """识别日志记录中的 access_token 明文值并脱敏(保留末 4 位)。

    覆盖三种序列化形态:
      1. JSON 字符串:  '"access_token": "VALUE"'   或  '"access_token":"VALUE"'
      2. Python repr "'access_token': 'VALUE'"   或   "access_token=VALUE"
      3. URL query    'access_token=VALUE&...'   或   'access_token=VALUE'

    挂载点settings.LOGGING.filters['access_token_mask'] →
            settings.LOGGING.handlers.{aliyun|console}.filters
    """

    # 正则模式说明plan 阶段确认 / 微调)
    _PATTERNS = [
        # 1) JSON 字符串:双引号
        re.compile(r'("access_token"\s*:\s*")([^"]+)(")'),
        # 2) Python dict repr单引号
        re.compile(r"('access_token'\s*:\s*')([^']+)(')"),
        # 3) URL query以 `&` / 空格 / 行尾为终止
        re.compile(r'(access_token=)([^&\s"\']+)'),
        # 4) 兜底:等号或冒号 + 空格分隔(容错 logger.info("access_token: VALUE") 风格)
        re.compile(r'(access_token\s*[:=]\s*)([^\s,;)\]\}"\']+)'),
    ]

    def _mask_in_text(self, text: str) -> str:
        if not isinstance(text, str) or 'access_token' not in text.lower():
            return text
        for pattern in self._PATTERNS:
            text = pattern.sub(lambda m: self._sub_mask(m), text)
        return text

    def _sub_mask(self, match) -> str:
        groups = match.groups()
        if len(groups) == 3:
            # 模式 1 / 2('"access_token":"', VALUE, '"')
            return groups[0] + mask_token(groups[1]) + groups[2]
        elif len(groups) == 2:
            # 模式 3 / 4('access_token=', VALUE)
            return groups[0] + mask_token(groups[1])
        return match.group(0)

    def filter(self, record: logging.LogRecord) -> bool:
        # 1. record.msg 字符串脱敏
        if isinstance(record.msg, str):
            record.msg = self._mask_in_text(record.msg)

        # 2. record.args 元组 / 字典中的元素脱敏
        if record.args:
            if isinstance(record.args, dict):
                record.args = {
                    k: (mask_token(v) if k == 'access_token' and isinstance(v, str)
                        else self._mask_in_text(v) if isinstance(v, str)
                        else v)
                    for k, v in record.args.items()
                }
            elif isinstance(record.args, tuple):
                record.args = tuple(
                    self._mask_in_text(a) if isinstance(a, str) else a
                    for a in record.args
                )

        return True   # filter 永远不丢弃 record

关键要点

Anti-Patterns to Avoid

  • 在 logger 名上挂 filterloggers.X.filtersfilter 挂在 logger 上仅过滤直接通过该 logger 的 record不过滤通过 handler 的;挂在 handler 上才统一覆盖所有 logger → handler 的链路。本 phase 必须挂在 handlers不挂在 loggers。
  • logging.Formatter 子类替代 FilterFormatter 是「格式化输出文本」阶段,对 record.args 是 dict / 复杂结构时改不动CONTEXT 也锁定走 Filter本注解仅作为复盘备查。
  • 在 view 内做 try/except 包裹脱敏view 应只关心业务,脱敏是横切关注点 —— 走 filter 不要往 view 里塞。
  • regex 写成贪婪 .+:会跨字段误吃。本 phase 4 个模式全部用 [^"]+ / [^']+ / [^&\s]+ 限定终止符。

Don't Hand-Roll

Problem Don't Build Use Instead Why
客户端凭据 GET serializer 新建 CredentialSlotClientSerializer 复用 aiapp.serializers.CredentialSlotSerializer CONTEXT 已锁定;字段集与管理端完全一致;分化会造成两处更新漂移
客户端 token 校验 新建 client-only auth class 复用 RedisTokenAuthentication get_user_id_from_token (utils.py:39-45) 已经先查 admin_token 再查 tokenadmin / user 都返回 user_idIsAuthenticated 对两种 token 都通过 —— 这正是 CONTEXT 决策「admin user 也是手机用户的超集」的实现基础
单例获取 新建 CredentialSlot.objects.get_or_create(pk=1) 复用 CredentialSlot.get_solo() Phase 1 已实现,单一入口
末 4 位脱敏 自己写正则切片 复用 common.utils.mask_token Phase 1 已实现,行为已被 Phase 2 验收
标准响应壳层 手搓 dict + Response 复用 success_response / error_response 与 Phase 2 一致
Swagger response 标准化 schema 手写 openapi.Schema 复用 get_standardized_response_schema Phase 2 admin view 已用drf-yasg 中间件兜层一致

关键洞察:本 phase 95% 的「实现」是「复用」。CRED-05 真正的新代码只有 view 类骨架(约 25-30 行)+ 1 行 url 注册。CRED-06 真正的新代码只有 filter 类(约 60-80 行)+ settings.LOGGING 4-5 行修改。

Runtime State Inventory

本 phase 不涉及 rename / refactor / migration —— 是纯新增 view + 新增 filter无需迁移现有数据 / 配置 / 注册项。

结论:跳过此节(无 runtime state 需要清点)。

Common Pitfalls

Pitfall 1filter 挂错位置loggers vs handlers

What goes wrongfilter 挂在 loggers['aiapp'].filters 而不是 handlers['aliyun'].filters,结果只过滤直接通过 getLogger('aiapp') 的 record过滤 root logger 或 django logger 走到同一 aliyun handler 的 record。 Why it happensPython logging filter 在 logger 和 handler 两个层级都能挂;新手不区分。 How to avoidCONTEXT 已明示「在 LOGGING.handlers.aliyun / console 等 handler 上配置 filters: ['access_token_mask']」。plan 必须把 filter 挂在 handlers 段,不要挂在 loggers 段。 Warning signsgrep 验证日志中含 'aiapp' 模块 logger 输出脱敏,但 'common''userapp' 模块的 access_token 输出仍是明文 → 说明 filter 漏挂。

Pitfall 2AliyunLogHandler.emitrecord.getMessage() 但不更新 record.msg

What goes wrongfilter 改了 record.msg,但若有人用 record.message 或在 filter 之前已经调用过 record.getMessage() 缓存了原始消息handler emit 时用的可能是旧值。 Why it happensrecord.messageFormatter.format() 调用 record.getMessage() 后赋值的属性。filter 在 handler 里执行时 record.message 还不存在handler emit 调用 record.getMessage() 会基于 record.msg + record.args 实时拼出 → 只要 filter 改了 record.msgrecord.argsemit 时拿到的就是脱敏后的版本。验证 common/aliyun_logging.py:23 用的就是 record.getMessage(),符合预期。 How to avoidfilter 同时改 record.msgrecord.args,不只改 record.msg。本 RESEARCH 的骨架已包含两者。 Warning signs:手动测试 logger.info("access_token=%s", "test_zzz") 时 console 仍输出 test_zzz 明文 → 说明只改了 record.msg 没改 record.args

Pitfall 3误把 user/admin Bearer token 当 access_token 脱敏

What goes wronguserapp/authentication.py:23logger.debug(f"Authorization header: {token}") 是 user/admin token30 天 Redis 凭证),不是 CredentialSlot.access_token。filter 的正则应匹配 access_token 字段名,不要泛化到 Authorization header:token: 这类格式 —— 那是另一类敏感数据,不在本 phase 范围。 Why it happens开发者一看到「token」字样就想脱敏全部。 How to avoidfilter 正则 4 个模式全部以 access_token 字段名为前缀锚点,不脱敏裸 token / Bearer / Authorization header。 Warning signs:用户在 Aliyun 日志里发现 Authorization header: 后面的 user-token 也被改成 *****xxxx —— 说明 filter 写得太宽,需要收紧。 Notelogger.debug 在 LOGGING 配置下走 'django' logger__name__='userapp.authentication' 命中 root 兜底root 默认 WARNING所以实际 DEBUG 不会被任何 handler 处理);该行当前在生产环境不输出。详见下文「Access Token 实际泄露路径」。

Pitfall 4access_token 字段值含 JSON 特殊字符未转义

What goes wrong:若 access_token 值含 "\,正则的 [^"]+ 会提前终止,匹配残缺。 Why it happensaccess_token 一般是 base64 / hex / 大小写字母数字混合,第三方服务商 (阿里云 / 火山 / 腾讯) 都不会在 token 中放 JSON 特殊字符;但理论上不严密。 How to avoid:本 phase 的 access_token 由阿里云 / 第三方 SDK 颁发,业界惯例是 alphanumeric + -_.~。可接受 4 个模式的正则覆盖。若未来出现含 " 的 tokenfilter 可降级为「字段值整体打 ***」处理。 Warning signs:日志中出现 "access_token": "abc\"def\"... 残缺脱敏。

Pitfall 5Django 4.2 dictConfig 工厂语法

What goes wrong:写成 'class': 'common.logging.filters.AccessTokenMaskFilter' 而不是 '()': 'common.logging.filters.AccessTokenMaskFilter'Why it happenshandler 用 'class',但 filter 用 '()'How to avoidfilter 段用 '()': 'dotted.path.to.FilterClass'(无参实例化);如要传参,再加 'arg1': value。本 phase 无参,骨架已写正确。 Warning signs:启动报 ValueError: Unable to configure filter 'access_token_mask' 或类似。

Code Examples

Example 1客户端 view 注册(完整改动 diff 等价物)

# Source: 1:1 复刻 aiapp/views.py:600-656 (CredentialSlotAdminView)
# 增量CredentialSlotClientView 类(约 25 行)+ 1 行 url 注册

# === 1) aiapp/views.py 末尾追加 ===
class CredentialSlotClientView(APIView):
    """通用凭据槽位客户端读取接口user/admin token 鉴权,明文返回)。"""
    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="读取成功")


# === 2) aiapp/views.py 中追加客户端 schema紧邻 _credential_slot_data_schema===
_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'),
    },
)


# === 3) qy_lty/urls.py imports 段追加 ===
from aiapp.views import CredentialSlotClientView

# === 4) qy_lty/urls.py api_urlpatterns 列表追加(在 common/upload/ 之后、v1/admin/ 之前) ===
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('credential-slot/', CredentialSlotClientView.as_view(), name='client_credential_slot'),  # ← 新增
    path('v1/admin/', include('userapp.admin_urls')),
]

Example 2filter 注册settings.LOGGING diff 等价物)

# Source: qy_lty/settings.py:372-412 + Django 4.2 logging dictConfig 标准

# 修改前settings.py:372-412 当前实测):
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},
    },
}

# 修改后:
LOGGING = {
    'version': 1,
    'disable_existing_loggers': False,
    '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},
    },
}

Example 3RedisTokenAuthentication 双 token 兼容性证实

# Source: userapp/utils.py:39-45 (verbatim)
def get_user_id_from_token(token):
    # 优先尝试从管理员token中获取
    user_id = cache.get(f"admin_token:{token}")
    if user_id is not None:
        return user_id
    # 再尝试从普通用户token中获取
    return cache.get(f"token:{token}")
# Source: userapp/authentication.py:14-34 (verbatim)
def authenticate(self, request):
    authorization = request.headers.get('Authorization')
    if not authorization:
        return None
    if len(authorization.split(' ')) < 2:
        return None
    token = authorization.split(' ')[1]
    if not token:
        return None
    logger.debug(f"Authorization header: {token}")          # ⚠ 见下文「实际泄露路径」#3 注

    user_id = get_user_id_from_token(token)                  # ← 双查 admin/useradmin 优先
    if not user_id:
        raise AuthenticationFailed('Invalid token')

    try:
        user = ParadiseUser.objects.get(id=user_id)
    except ParadiseUser.DoesNotExist:
        raise AuthenticationFailed('User not found')

    return (user, None)                                      # ← 返回真实 ParadiseUser 对象

结论证实 CONTEXT 「访问无 admin 限制」的 token 决策

  • admin token 与 user token 都返回真实的 ParadiseUser 对象
  • admin user 一般 is_staff=True,普通 user is_staff=False,但 IsAuthenticated 检查的是 request.user.is_authenticated(继承自 AbstractUser,注册过的 user 永远 True检查 is_staff
  • 因此 permission_classes = [IsAuthenticated] 对两种 token 都通过 ✓ —— 这就是 CONTEXT「都允许」决策的实现基础
  • Phase 2 admin view 用 _ensure_admin(request) 显式检查 request.user.is_staff 来拒绝 user tokenPhase 3 客户端 view 直接不调 _ensure_admin,达到「都允许」效果

State of the Art

Old Approach Current Approach When Changed Impact
logging.Filter dictConfig Python 3.x stdlib filter 是 Python logging 模块自 Python 2.x 起就有的标准能力Django 4.2 LOGGING dictConfig 完全支持;本 phase 不依赖新特性
自己写 access log 中间件 DRF + StandardResponseMiddleware(不打 body 日志) 项目已有 已确认中间件不打日志,无需改

Deprecated/outdated:本 phase 无 deprecated 项 —— 全部走 Python stdlib + Django 4.2 + 已有项目工具链。

Assumptions Log

本 RESEARCH 所有结论均通过 read 实证仓库源码或 grep 验证,无需用户确认任何假设。

若 plan 阶段发现以下「实证结论」不符合用户预期,再回到 discuss-phase 修正:

# Claim Section Risk if Wrong
A1 RedisTokenAuthentication 对 admin / user token 都返回 is_authenticated=True 的 user 对象 Code Examples Ex.3 若 risk 为真admin token 持有者无法访问 client view需要在 client view 加显式分支 —— 但这违反 CONTEXT「都允许」决策。已通过 read userapp/utils.py + authentication.py + Phase 2 验收记录证实,无 risk
A2 StandardResponseMiddleware.process_response 不调用任何 logger.xxx(即不打日志) access_token 实际泄露路径 + Pitfall 复盘 若 risk 为真:会有第 4 条隐蔽泄露路径需要覆盖 —— 已通过 grep logger\.[a-z]+ in common/middleware.py 验证0 命中,无 risk
A3 仓库现有 view 中没有任何代码 logger.xxx(... CredentialSlot.access_token ...) access_token 实际泄露路径 若 risk 为真:会有具体 view 内 logger 泄露需要单独修 —— 已通过 grep logger\..*token 全仓验证:仅 user/admin tokenBearer和 RTC token无 CredentialSlot.access_token 字段,无 risk

Open Questions

无 open questions —— 本 phase 所有决策均已被 CONTEXT 锁定 + 本 RESEARCH 实证。

Environment Availability

本 phase 是纯 Python 代码改动,无外部工具 / 服务依赖,跳过此节。

Aliyun 日志服务在生产已可用 —— 见 common/aliyun_logging.py 已稳定运行Redis/PostgreSQL 与 Phase 1/2 共享,已可用。)

Validation Architecture

检查 qy_lty/.planning/config.json 是否设置 workflow.nyquist_validation: false —— 实地无 .planning/config.json 文件按缺省「enabled」处理。

Test Framework

Property Value
Framework 当前仓库 未配置 pytest / pytest-django候选优先级 #5brownfield 候选项已记录)尚未消化。本 phase 沿用 Phase 2 同款验收方式:django.test.Client 程序化端到端测试in-process不启 daphne
Config file 无(每个 phase 临时写 _phaseN_verify.py 跑完即删,参考 Phase 2 的 _phase2_verify.py 模式)
Quick run command python manage.py shell < _phase3_verify.pyplan 阶段创建脚本)
Full suite command 同上

Phase Requirements → Test Map

Req ID Behavior Test Type Automated Command File Exists?
CRED-05 client GET 携 user token → 200 + 明文 access_token integration django.test.Client.get('/api/credential-slot/', HTTP_AUTHORIZATION=f'Bearer {user_token}') Wave 0plan 阶段写 _phase3_verify.py
CRED-05 client GET 携 admin token → 200 + 明文(不区分) integration 同上换 admin_token Wave 0
CRED-05 client GET 无 token → 401 + 标准壳层 integration Client.get('/api/credential-slot/') Wave 0
CRED-05 client GET 过期 token → 401 integration Client.get(... HTTP_AUTHORIZATION='Bearer fake_token_zzz') Wave 0
CRED-05 swagger 暴露 /api/credential-slot/ 路径 integration Client.get('/swagger.json/') + path key 校验 Wave 0
CRED-05 端到端往返admin PUT roundtrip → client GET 拿到一致明文 integration 串联 PUT + GET Wave 0
CRED-06 filter 单元:logger.info("access_token=secret_zzz")record.msg 已脱敏 unit pytest-free 自检:实例化 filter构造 LogRecordfilter()assert record.getMessage()*****zzz 不含 secret_zzz Wave 0
CRED-06 filter 4 种模式覆盖JSON / Python repr / query / 等号兜底 unit 同上分 4 case Wave 0
CRED-06 日志链路真实输出:捕获 stderr触发 PUT /api/v1/admin/credential-slot/,断言 stderr 不含完整明文 integration Client.put(...) + capsys / StringIO redirect Wave 0

Sampling Rate

  • Per task commitpython manage.py shell < _phase3_verify.py 全量跑一次
  • Per wave merge:同上
  • Phase gateCRED-05 的 6 项 client view + CRED-06 的 3 项 filter 全 PASS

Wave 0 Gaps

  • _phase3_verify.py仓库根plan 阶段写、phase end 删)—— 端到端 client view + filter 行为校验
  • CredentialSlot DB 终态:以 Phase 2 02-VERIFICATION.md 末尾「DB 终态记录」(probe_app / probe_secret_xxxx) 为起点Phase 3 验收前后 DB 状态保持不变(验收脚本主动还原)
  • 无独立 test framework 引入 —— 与 Phase 2 一致沿用 django.test.Client 模式

Security Domain

检查 qy_lty/.planning/config.json 是否设置 security_enforcement: false —— 实地无 .planning/config.json 文件按缺省「enabled」处理。

Applicable ASVS Categories

ASVS Category Applies Standard Control
V2 Authentication yes RedisTokenAuthentication已有TTL 30 天 Redis 凭证)
V3 Session Management partial Token 即 session 替代30 天 TTL本 phase 不引入新 session 机制
V4 Access Control yes IsAuthenticatedclient view/ _ensure_adminadmin view 已落地)
V5 Input Validation n/a client view 仅 GET无请求 body
V6 Cryptography n/a 不在本 phaseDB at-rest 加密已 deferred
V7 Error Handling and Logging yes 本 phase 核心AccessTokenMaskFilter 防止敏感字段进 Aliyun 日志服务
V9 Communications partial HTTPS 由反向代理Nginx兜底不在本 phase
V12 API and Web Service yes DRF + drf-yasg + Bearer token已落地

Known Threat Patterns for Django + DRF + Aliyun Log Service

Pattern STRIDE Standard Mitigation
logger.info(f"PUT body: {request.data}") 类型代码导致 access_token 进生产日志 Information Disclosure 本 phase AccessTokenMaskFilter 在 handler 层兜底,覆盖未来开发者的不慎 logger 调用
Bearer token 在 Authorization header: 日志被打印 Information Disclosure 不在本 phase 范围CRED-06 仅针对 access_token 字段user/admin Bearer token 是另一类敏感数据,留 v2.x 候选优先级 #1 / #3 处理)
客户端 view 被 admin token 调用 Privilege Escalationfalse positive 不构成提权 —— 客户端 view 返回的数据是「明文 APP ID + Access Token」admin 通过管理端 PUT 接口本来就能写明文所以读明文不是新增信息CONTEXT 已论证)
swagger 暴露 client endpoint 给未鉴权读者 Information Disclosure swagger 本身仅暴露 schema 不暴露数据schema 中 description 标注「明文,需 token」即可与 admin view 同模式)

Sources

Primary (HIGH confidence)

  • qy_lty/qy_lty/settings.py:372-412 —— LOGGING 配置全貌read 实证)
  • qy_lty/qy_lty/urls.py:49-60 —— api_urlpatterns 完整列表read 实证)
  • qy_lty/aiapp/views.py:600-687 —— CredentialSlotAdminView 完整模板read 实证)
  • qy_lty/aiapp/views.py:1-19 —— imports 段(确认 RedisTokenAuthentication / IsAuthenticated / mask_token / success_response / get_standardized_response_schema 均已可用)
  • qy_lty/aiapp/urls.py:1-13 —— 现有 ai 子 urlpatternsread 实证)
  • qy_lty/userapp/admin_urls.py:1-15 —— admin 命名空间注册read 实证 + Phase 2 模板)
  • qy_lty/userapp/urls.py:1-32 —— 客户端命名空间风格参考read 实证)
  • qy_lty/userapp/authentication.py:1-34 —— RedisTokenAuthentication + logger.debug(f"Authorization header: {token}") 行号 23read 实证)
  • qy_lty/userapp/utils.py:39-45 —— get_user_id_from_token 双查逻辑read 实证)
  • qy_lty/common/middleware.py:1-145 —— StandardResponseMiddleware 全文read 实证0 处 logger 调用)
  • qy_lty/common/utils.py:10-32 —— mask_token 行为契约read 实证)
  • qy_lty/common/responses.py:41-52 —— success_response 行为read 实证)
  • qy_lty/common/aliyun_logging.py:16-36 —— AliyunLogHandlerread 实证:用 record.getMessage()filter 改 record.msg 后即生效)
  • qy_lty/.planning/phases/02-admin-rest/02-VERIFICATION.md —— Phase 2 验收记录 + DB 终态read 实证)

Secondary (MEDIUM confidence)

  • 无 —— 本 phase 不依赖外部文档,所有结论均通过仓库源码 read 实证

Tertiary (LOW confidence)

  • 无 —— 不依赖 web search

CITED 文档参考(按需查阅,非本 phase 阻塞项)

Metadata

Confidence breakdown:

  • Standard stack: HIGH —— 全部已存在于 requirements.txt本 phase 不引入新依赖;版本沿用生产已运行版本
  • Architecture: HIGH —— 1:1 复刻 Phase 2 admin view 模板CRED-06 的 LOGGING.filters 是 Django dictConfig 标准
  • Pitfalls: HIGH —— 通过实证仓库现状middleware 不打日志 / view 不打 access_token / filter 注册位置)已逐项验证

Research date: 2026-05-08 Valid until: 2026-06-0830 天,因仓库代码相对稳定 + 本 phase 是纯增量改动;超期前若 Phase 1/2 文件被改动,需重新 read 验证)


附录access_token 实际泄露路径详细分析

CONTEXT 假设的 3 条路径 vs 实际证据

# CONTEXT 假设的路径 实际是否泄露 证据
1 管理端 PUT 请求体(access_token 进日志) ✗ 不会自动泄露 StandardResponseMiddleware.process_responseprocess_exception 不调任何 loggergrep logger\. in common/middleware.py = 0 命中Django 默认 access logdjango.request logger只在 ERROR 级别settings.py:391-394 配置 ERROR + console only输出 stack trace不输出请求 body除非未来开发者在 CredentialSlotAdminView.put 内显式 logger.info(f"PUT body: {request.data}"),否则 PUT 请求体永不进 logger
2 管理端 GET 响应体(access_token 进日志) ✗ 不会泄露 同上 + Phase 2 admin view GET 已经在 view 层 _build_response_datamask_token 替换 access_token 字段;即使开发者后续加 logger.info(success_response.data),也只会打到末 4 位脱敏值。真正的明文 access_token 仅在 serializer.save() 的 instance 内存对象中,未走 logger 路径
3 客户端 GET 响应体(access_token 进日志) ✗ 不会自动泄露 客户端 view 返回 success_response(data=serializer.data)serializer.data 的明文 access_token 在 Response 对象内,被 StandardResponseMiddleware 包裹后写 response.content,但不进 logger;同 #1除非开发者显式 logger 调用,否则不泄露

实际确实存在的泄露代码位置

经全仓 grep logger\.[a-z]+.*token 验证,真实存在的 token 进 logger 代码点:

文件:行号 代码 影响范围 是否本 phase 范围
userapp/authentication.py:23 logger.debug(f"Authorization header: {token}") 每次 携 Bearer token 的 HTTP 请求都会 debug-log Bearer token 明文 ⚠ 范围外但相邻:这是 user/admin 登录 token30 天 Redis 凭证),不是 CredentialSlot.access_token。当前 LOGGING 配置 userapp logger level=INFOsettings.py:406-410DEBUG 级别不会被任何 handler 处理,故生产环境不实际输出。但若未来有人改成 logger.info 或调高 logger level会立刻泄露 30 天 Redis token
device_interaction/auth.py:47 print(f"Token authentication error: {str(e)}") WebSocket 鉴权异常时 print 范围外:是 exception message 不含 token 值
device_interaction/views.py:1215 logger.error(f"Failed to get token by MAC address: {str(e)}") RTC token 失败 path 范围外exception message 不含 token
device_interaction/consumers.py:23/27/75/103 4 处 logger.warning/error 关于 token authentication failure WebSocket 鉴权失败路径 范围外exception 不含 token 值
aiapp/audio/AliyunAudioService.py:48 print("token = " + token) NLS 调用 stdout print ⚠ 是阿里云 NLS appkey/token与 access_token 字段同名但值不同AccessTokenMaskFilter 的字段名匹配会误伤这条 print —— 不过 print 不走 loggerfilter 不影响 print。stdout 在容器中可能进日志 → 留 v2.x 处理
aiapp/audio/test.py:38 同上 测试代码 范围外:不在生产路径

结论CRED-06 的真实价值

AccessTokenMaskFilter当前仓库状态下几乎没有 record 可过滤 —— 真正的 CredentialSlot.access_token 在生产代码中根本没有进 logger 的路径。

但这恰是 CRED-06 的价值所在 —— 防御性兜底

  1. 任何后续开发者写 logger.info(f"PUT body: {request.data}") 类型代码 → 立即被 filter 兜住
  2. Django 默认 access log 若被升级为 INFO 级别(如调试需求) → filter 兜住
  3. Phase 2 / 3 之外的 view 若需要 dump 整个 model → filter 兜住

强烈建议 plan 阶段:在 CRED-06 task 内显式新增一个验证步骤 ——

  • 触发一次 PUT /api/v1/admin/credential-slot/(用 Phase 2 admin view临时 在 admin view 末尾加一行 logger.info(f"PUT 请求体测试: {request.data}")
  • 验证 stderr / Aliyun 日志该字段已脱敏 →
  • 删掉这行临时 logger.infoCRED-06 验收完毕)。

这样能验证 filter 真实在工作,而不是空跑。


Phase 3 RESEARCH 完成;下一步 plan-phase 可基于本文件创建任务清单