29 KiB
29 KiB
phase, plan, type, wave, depends_on, files_modified, autonomous, requirements, must_haves
| phase | plan | type | wave | depends_on | files_modified | autonomous | requirements | must_haves | ||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| 02-admin-rest | 01 | execute | 1 |
|
true |
|
|
Purpose：
- CRED-03：管理后台读取脱敏后的凭据槽位（仅末 4 位明文，避免运营在 admin UI 看到完整明文）
- CRED-04：管理后台以全字段覆写方式更新凭据槽位，PUT 响应同样走脱敏避免明文回显
Output：
- 新增
CredentialSlotSerializer（ModelSerializer，3 字段 + read_only_fields + extra_kwargs allow_blank） - 新增
CredentialSlotAdminView（自定义 APIView + 手写 GET/PUT，1:1 复刻 RTCChatHistoryAPIView 风格） - 在
userapp/admin_urls.py追加path('credential-slot/', ...)注册 - View 方法挂
@swagger_auto_schema装饰器，响应 schema 显式标注 access_token 末 4 位脱敏掩码语义
<execution_context> @$HOME/.claude/get-shit-done/workflows/execute-plan.md @$HOME/.claude/get-shit-done/templates/summary.md </execution_context>
@.planning/PROJECT.md @.planning/ROADMAP.md @.planning/STATE.md @.planning/REQUIREMENTS.md @.planning/phases/02-admin-rest/02-CONTEXT.md @.planning/phases/02-admin-rest/02-RESEARCH.mdPhase 1 落地决策（为何模型走 pk=1 + get_solo + mask_token）
@.planning/phases/01-credential-data-layer/01-01-SUMMARY.md @.planning/phases/01-credential-data-layer/01-02-SUMMARY.md
1:1 复刻样板（必读）
@qy_lty/aiapp/views.py @qy_lty/aiapp/serializers.py @qy_lty/aiapp/models.py @qy_lty/aiapp/urls.py @qy_lty/userapp/views.py @qy_lty/userapp/admin_urls.py @qy_lty/userapp/authentication.py @qy_lty/userapp/utils.py @qy_lty/common/responses.py @qy_lty/common/middleware.py @qy_lty/common/swagger_utils.py @qy_lty/common/utils.py @qy_lty/qy_lty/urls.py
【已存在 — 复用】Model: qy_lty/aiapp/models.py
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)
def save(self, *args, **kwargs):
# pk=1 单例钩子：已有记录时把 pk 重定向，DB 永远只有一行
...
@classmethod
def get_solo(cls):
instance, _created = cls.objects.get_or_create(pk=1)
return instance
【已存在 — 复用】Auth: qy_lty/userapp/authentication.py
class RedisTokenAuthentication(BaseAuthentication):
# 读取 Authorization: Bearer <token>，调 get_user_id_from_token：
# 优先查 admin_token:{token}，否则 token:{token}；都查不到 → AuthenticationFailed (401)
# 命中 → 返回 (ParadiseUser, None)
def authenticate(self, request): ...
注意：本类不区分 admin / user token。区分点是 request.user.is_staff（只有走 AdminEmailLoginView 的用户才会是 staff）。
【已存在 — 复用】Helpers: qy_lty/common/responses.py
def success_response(data=None, message="操作成功", code=200, **kwargs) -> Response # 200
def error_response(message="操作失败", code=400, status_code=400, **kwargs) -> Response
两者已构造壳层四字段（success / code / message / data），与 StandardResponseMiddleware 协同（middleware.py:53-55 检查 success+code 二者皆在则不二次包装）。
【已存在 — 复用】Tool: qy_lty/common/utils.py
def mask_token(token: str, visible_tail: int = 4, mask_char: str = '*') -> str:
# 'sk-abcdef1234' -> '*********1234'
# '' -> ''
# 'abc' -> '***' # 短于 visible_tail 时全脱敏
【已存在 — 复用】Swagger: qy_lty/common/swagger_utils.py
def get_standardized_response_schema(data_schema=None) -> openapi.Schema
# 返回 OpenAPI Schema（type=OBJECT，properties=success/code/message + 可选 data）
【新增 — 本 plan 落地】Serializer: qy_lty/aiapp/serializers.py
class CredentialSlotSerializer(serializers.ModelSerializer):
class Meta:
model = CredentialSlot
fields = ['app_id', 'access_token', 'updated_at']
read_only_fields = ['updated_at']
extra_kwargs = {
'app_id': {'allow_blank': True, 'allow_null': False, 'required': False},
'access_token': {'allow_blank': True, 'allow_null': False, 'required': False},
}
【新增 — 本 plan 落地】View: qy_lty/aiapp/views.py 末尾
class CredentialSlotAdminView(APIView):
authentication_classes = [RedisTokenAuthentication]
permission_classes = [IsAuthenticated]
def get(self, request): ... # 返回脱敏 data
def put(self, request): ... # 全字段覆写 + 脱敏响应
完整修改后文件内容（直接覆写）：
from rest_framework import serializers
from .models import ChatMessage, CredentialSlot
class ChatMessageSerializer(serializers.ModelSerializer):
class Meta:
model = ChatMessage
fields = ['id', 'user', 'bot', 'message', 'timestamp', 'sender', 'message_type', 'message_audio_url', 'message_video_url']
read_only_fields = ['id', 'timestamp', 'sender']
class CredentialSlotSerializer(serializers.ModelSerializer):
"""通用凭据槽位序列化器（明文存储，脱敏由 view 层完成）。
设计动机（per CONTEXT.md D-Serializer）：
- 脱敏放 view 层不放 serializer：PUT 路径需要明文走 is_valid + save，serializer
不应承担"既要明文又要脱敏"的双重责任。
- app_id / access_token 在模型层 blank=True, default=''，对应 serializer 配
allow_blank=True, allow_null=False, required=False；既允许空字符串覆写、又
拒绝 None；缺字段时由 ModelSerializer 默认行为（用现有值兜底）。
- updated_at 由模型层 auto_now=True 自动维护，read_only 双重保险。
"""
class Meta:
model = CredentialSlot
fields = ['app_id', 'access_token', 'updated_at']
read_only_fields = ['updated_at']
extra_kwargs = {
'app_id': {'allow_blank': True, 'allow_null': False, 'required': False},
'access_token': {'allow_blank': True, 'allow_null': False, 'required': False},
}
注意：
- 不要发明
CredentialSlotReadSerializer/CredentialSlotWriteSerializer拆分两个类（CONTEXT.md / RESEARCH.md "Alternatives Considered" 已锁定单一 serializer + view 层脱敏） - 不要在 serializer 内写
to_representation覆写做脱敏（同上锁定） - 不要给
app_id/access_token加allow_null=True（与模型层 blank=True/default='' 不一致） cd qy_lty && python -c "from aiapp.serializers import CredentialSlotSerializer; from aiapp.models import CredentialSlot; s = CredentialSlotSerializer(CredentialSlot.get_solo()); assert set(s.data.keys()) == {'app_id', 'access_token', 'updated_at'}, s.data.keys(); print('OK fields=', list(s.data.keys()))" <acceptance_criteria>python -c "from aiapp.serializers import CredentialSlotSerializer"无 ImportError- serializer 实例化已有的 pk=1 记录后
.data三键齐全：app_id/access_token/updated_at - serializer.fields['updated_at'].read_only == True
- serializer.fields['app_id'].allow_blank == True 且 allow_null == False
- serializer.fields['access_token'].allow_blank == True 且 allow_null == False
- 文件不再含其它 serializer 类（仅追加 CredentialSlotSerializer）
</acceptance_criteria>
aiapp/serializers.py含 CredentialSlotSerializer 类，三字段对齐模型，updated_atread-only；GET 用它返回 .data 含明文（view 层稍后脱敏），PUT 用它做 is_valid + save。
修改第 5 行（追加 CredentialSlot）：
# 修改前：
from .models import ChatMessage, Bot
# 修改后：
from .models import ChatMessage, Bot, CredentialSlot
修改第 7 行（追加 CredentialSlotSerializer）：
# 修改前：
from .serializers import ChatMessageSerializer
# 修改后：
from .serializers import ChatMessageSerializer, CredentialSlotSerializer
在第 13 行（from drf_yasg.utils import swagger_auto_schema 已存在）之后追加两行（如果文件中已存在则跳过该行；只追加缺失的）：
from common.utils import mask_token
from common.swagger_utils import get_standardized_response_schema
Step 2：在 qy_lty/aiapp/views.py 文件末尾（紧跟 RTCChatHistoryAPIView.delete 之后）追加完整代码块：
# ======================================================================
# Phase 2 — 通用凭据槽位管理端读写接口（CRED-03 + CRED-04）
# 1:1 复刻 RTCChatHistoryAPIView 的单 URL 多方法 APIView 风格
# ======================================================================
class CredentialSlotPutRequestSchema(serializers.Serializer):
"""drf-yasg 专用 — PUT 请求体 schema（不参与实际写入校验，仅给 swagger 看）。
实际写入校验由 CredentialSlotSerializer.is_valid 执行。
"""
app_id = serializers.CharField(
required=False, allow_blank=True,
help_text="第三方服务商分配的 APP ID（明文写入；缺省时保留原值）"
)
access_token = serializers.CharField(
required=False, allow_blank=True,
help_text="第三方服务商访问令牌（明文写入；响应阶段会脱敏返回末 4 位）"
)
# 响应 data 子 schema：access_token 字段 description 显式标注脱敏掩码语义
_credential_slot_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 位脱敏掩码（如 "*********1234"，前缀字符数 = 原长 - 4）',
),
'updated_at': openapi.Schema(
type=openapi.TYPE_STRING,
format='date-time',
description='最近一次更新时间（ISO 8601）',
),
},
)
class CredentialSlotAdminView(APIView):
"""通用凭据槽位管理端读写接口（admin token 鉴权）。
GET: 返回 access_token 末 4 位脱敏后的凭据槽位
PUT: 全字段覆写凭据槽位（空记录场景自动 get_or_create），响应同样脱敏
"""
authentication_classes = [RedisTokenAuthentication]
permission_classes = [IsAuthenticated]
tags = ['通用凭据槽位（管理端）']
def _ensure_admin(self, request):
"""admin-only 二次校验：拒绝非 staff 用户（含普通 user token 持有者）。
per RESEARCH.md：仓库零处 IsAdminTokenAuthenticated permission 类；
现有 AdminEmailLoginView (userapp/views.py:748-754) / AdminLogoutView 一律走
视图内 is_staff 检查。统一沿用此模式，不发明新 permission 类。
"""
if not request.user.is_staff:
logger.warning(
f"Non-admin user attempted CredentialSlot admin endpoint: user_id={request.user.id}"
)
return error_response(
message="需要管理员权限",
code=403,
status_code=status.HTTP_403_FORBIDDEN,
)
return None
def _build_response_data(self, instance):
"""构造脱敏后的响应 data 字典。
per CONTEXT.md：GET 与 PUT 响应都必须脱敏 access_token，避免运营在
admin UI 看到自己刚提交的明文回显（CONTEXT.md 决策"PUT 响应也走脱敏"）。
"""
serializer = CredentialSlotSerializer(instance)
data = dict(serializer.data)
# 关键脱敏点：用 instance.access_token（明文）走 mask_token，覆盖 serializer.data 里的明文
data['access_token'] = mask_token(instance.access_token)
return data
@swagger_auto_schema(
operation_description="读取通用凭据槽位（access_token 末 4 位脱敏返回，admin token 鉴权）",
responses={
200: openapi.Response('读取成功', get_standardized_response_schema(_credential_slot_data_schema)),
401: openapi.Response('未提供有效 token', get_standardized_response_schema()),
403: openapi.Response('需要管理员权限', get_standardized_response_schema()),
},
security=[{'Bearer': []}],
tags=['通用凭据槽位（管理端）'],
)
def get(self, request):
forbidden = self._ensure_admin(request)
if forbidden:
return forbidden
instance = CredentialSlot.get_solo()
data = self._build_response_data(instance)
return success_response(data=data, message="读取成功")
@swagger_auto_schema(
request_body=CredentialSlotPutRequestSchema,
operation_description="全字段覆写通用凭据槽位（admin token 鉴权；写入后响应脱敏返回）",
responses={
200: openapi.Response('更新成功', get_standardized_response_schema(_credential_slot_data_schema)),
400: openapi.Response('参数无效', get_standardized_response_schema()),
401: openapi.Response('未提供有效 token', get_standardized_response_schema()),
403: openapi.Response('需要管理员权限', get_standardized_response_schema()),
},
security=[{'Bearer': []}],
tags=['通用凭据槽位（管理端）'],
)
def put(self, request):
forbidden = self._ensure_admin(request)
if forbidden:
return forbidden
instance = CredentialSlot.get_solo() # 空记录场景自动 get_or_create
serializer = CredentialSlotSerializer(instance, data=request.data)
if not serializer.is_valid():
return error_response(
message="参数无效",
code=400,
status_code=status.HTTP_400_BAD_REQUEST,
data=serializer.errors,
)
serializer.save() # auto_now 自动刷 updated_at
# 重新读取 instance.access_token：serializer.save() 后 instance 已被同步刷新；
# _build_response_data 内部会再次 dict(serializer.data) 拿最新 OrderedDict。
data = self._build_response_data(instance)
return success_response(data=data, message="凭据已更新")
关键约束（不要违反）：
- 不要把 view 写成
RetrieveUpdateAPIView子类（仓库零先例，per RESEARCH.md "Alternatives Considered"） - 不要直接
return Response({...}, status=200)；一律走success_response/error_response（避免 middleware 二次包装的不确定行为；per Pitfall 2） - 不要在 PUT 路径忘记脱敏：
return success_response(data=serializer.data)直接返回是 BUG ——serializer.data含明文 access_token（per Pitfall 3） - 不要新增
IsAdminTokenAuthenticatedpermission 类（仓库零先例，与现有约定相悖；per RESEARCH.md "Anti-Patterns"） - 不要把
_ensure_admin改成permission_classes = [IsAdminUser]（DRF 自带 IsAdminUser 也是 is_staff，但与本仓库"视图内手写 is_staff"统一约定不一致） cd qy_lty && python -c "from aiapp.views import CredentialSlotAdminView; v = CredentialSlotAdminView(); assert hasattr(v, 'get') and hasattr(v, 'put'); from userapp.authentication import RedisTokenAuthentication; assert RedisTokenAuthentication in CredentialSlotAdminView.authentication_classes; print('OK view loaded with auth')" <acceptance_criteria>python -c "from aiapp.views import CredentialSlotAdminView"无 ImportError- View 类拥有
get/put/_ensure_admin/_build_response_data方法 CredentialSlotAdminView.authentication_classes含RedisTokenAuthenticationCredentialSlotAdminView.permission_classes含IsAuthenticated（不是 IsAdminUser、不是自定义 admin permission）- grep
qy_lty/aiapp/views.py含mask_token(instance.access_token)至少一次（脱敏调用点） - grep
qy_lty/aiapp/views.py含if not request.user.is_staff至少一次（admin 二次校验） - grep
qy_lty/aiapp/views.py不含RetrieveUpdateAPIView（不准走 DRF 通用 view） - grep
qy_lty/aiapp/views.py不含IsAdminTokenAuthenticated（不准发明新 permission 类） - swagger 装饰器在 GET 与 PUT 各挂一份（grep
@swagger_auto_schema出现次数比改动前多 2） </acceptance_criteria>aiapp/views.py末尾追加CredentialSlotAdminView类（含 GET/PUT 两方法 + admin 二次校验 + 脱敏 helper）；顶部 import 已对齐；view 可被 import 不报错。
from django.urls import path
from .views import AdminEmailLoginView, AdminLogoutView
# Phase 2 — 通用凭据槽位管理端读写接口（CRED-03 + CRED-04）
from aiapp.views import CredentialSlotAdminView
# 管理员专用API路径
urlpatterns = [
# 管理员登录
path('login/', AdminEmailLoginView.as_view(), name='admin_login'),
# 管理员登出
path('logout/', AdminLogoutView.as_view(), name='admin_logout'),
# 通用凭据槽位（GET 脱敏读取 / PUT 全字段覆写；admin token 鉴权）
path('credential-slot/', CredentialSlotAdminView.as_view(), name='admin_credential_slot'),
# 后续可以添加更多管理员专用接口
]
关键约束：
- import 必须从
aiapp.views引入（凭据槽位 view 落地于 aiapp，不是 userapp） - 路径必须是
'credential-slot/'（trailing slash + 中划线，对齐 CONTEXT.md "trailing slash 沿用 Django 默认风格"） - name 必须是
'admin_credential_slot'（reverse 时使用） - 不要在
aiapp/urls.py重复注册（CONTEXT.md 锁定路由汇总点是 userapp/admin_urls.py） cd qy_lty && python -c "from django.urls import reverse; from django.conf import settings; import django; django.setup() if not settings.configured else None; from django.urls import get_resolver; r = get_resolver(); url = reverse('admin_credential_slot'); assert url == '/api/v1/admin/credential-slot/', f'got {url}'; print('OK url=', url)" 2>&1 || (cd qy_lty && DJANGO_SETTINGS_MODULE=qy_lty.settings python -c "import django; django.setup(); from django.urls import reverse; print('url=', reverse('admin_credential_slot'))") <acceptance_criteria>python manage.py check无 URL 注册类报错reverse('admin_credential_slot')返回/api/v1/admin/credential-slot/qy_lty/userapp/admin_urls.py含from aiapp.views import CredentialSlotAdminViewqy_lty/userapp/admin_urls.py不含将凭据槽位重复注册到aiapp/urls.py的逻辑qy_lty/aiapp/urls.py内未新增 credential-slot 注册（汇总点单一） </acceptance_criteria>/api/v1/admin/credential-slot/URL 可被 reverse 解析到CredentialSlotAdminView；Django check 通过；与 login/logout 在同一 admin namespace 注册块内。
- import 链路完整：
python -c "from aiapp.views import CredentialSlotAdminView; from aiapp.serializers import CredentialSlotSerializer"同行无 ImportError - URL 路由生效：
python manage.py shell -c "from django.urls import reverse; print(reverse('admin_credential_slot'))"输出/api/v1/admin/credential-slot/ - Django check 通过：
python manage.py check无 ERRORS / WARNINGS（W 级 noise 可忽略） - Swagger schema 暴露（运行时验证留 02-02）：本 plan 不启动 daphne，但通过 import 检查保证
@swagger_auto_schema装饰器无语法错误
reachability self-check（goal-backward）：
- truth #1（GET 脱敏 200）→ artifact: views.py CredentialSlotAdminView.get + serializers.py CredentialSlotSerializer + admin_urls.py path → reachable ✓
- truth #2（PUT 全字段覆写 + updated_at 刷新 + 响应脱敏）→ artifact: views.py CredentialSlotAdminView.put（serializer.save + _build_response_data）→ reachable ✓
- truth #3（PUT 在空记录场景 get_or_create）→ artifact: CredentialSlot.get_solo()（Phase 1 落地，本 plan 调用）→ reachable ✓
- truth #4（无 token → 401）→ artifact: RedisTokenAuthentication（Phase 0 已在）+ DRF NotAuthenticated → reachable ✓
- truth #5（user token → 403）→ artifact: views.py
_ensure_adminis_staff 校验 → reachable ✓ - truth #6（swagger 路径条目 + access_token 脱敏 description）→ artifact: views.py @swagger_auto_schema + _credential_slot_data_schema description → reachable ✓
<threat_model>
Trust Boundaries
| Boundary | Description |
|---|---|
| 公网 → /api/v1/admin/credential-slot/ | 来自 qy-lty-admin Web UI 的 HTTP 请求；untrusted Authorization header；untrusted PUT body |
| view → DB（CredentialSlot 表） | 内部信任边界；写入前必须经过 serializer.is_valid + admin 校验 |
| view → 响应回 admin 端 | 出站脱敏边界；access_token 必须经 mask_token 才能离开后端 |
STRIDE Threat Register
| Threat ID | Category | Component | Disposition | Mitigation Plan |
|---|---|---|---|---|
| T-02-01 | Spoofing | Authorization header | mitigate | RedisTokenAuthentication 走 Redis admin_token:{token} key 验证；TTL 30 天；token 验证失败抛 AuthenticationFailed → 401 |
| T-02-02 | Elevation of Privilege | user token 持有者调 admin 端点 | mitigate | view 内 _ensure_admin 早返回 403；CONTEXT.md success #5 的核心防线 |
| T-02-03 | Information Disclosure | GET 响应明文 access_token | mitigate | _build_response_data 强制走 mask_token(instance.access_token)；serializer 不在 to_representation 内做脱敏避免责任分散 |
| T-02-04 | Information Disclosure | PUT 响应回显明文 access_token | mitigate | PUT 路径 save() 后同样调 _build_response_data；不直接 return success_response(data=serializer.data) — 该写法会回显明文（per Pitfall 3） |
| T-02-05 | Tampering | mass assignment（未声明字段） | mitigate | DRF ModelSerializer 默认拒绝未在 Meta.fields 声明的字段；fields=['app_id', 'access_token', 'updated_at'] 三字段封死；updated_at read_only 双重保险 |
| T-02-06 | Tampering | PUT 重放 / 幂等性 | accept | 全字段覆写本身幂等（重放结果一致）；不需要 ETag / If-Match |
| T-02-07 | Denial of Service | 暴力 PUT 消耗 DB | accept | 现有架构无限流（PROJECT.md candidate priorities #2 跟踪）；本 phase 不引入新依赖；deferred ideas 已明确 |
| T-02-08 | Information Disclosure | access_token 写入 access log（请求体 / 响应体） | accept（本 phase）+ transfer（Phase 3） | 本 phase 仅做响应脱敏；阿里云 access log 链路过滤由 Phase 3 落地（CRED-06）；Phase 3 完成后此项关闭 |
| </threat_model> |
<success_criteria> 本 plan 落地完成的标志（plan 内自验）：
qy_lty/aiapp/serializers.py顶部from .models import ChatMessage, CredentialSlot，文件内含class CredentialSlotSerializer(serializers.ModelSerializer):qy_lty/aiapp/views.py顶部 import 区含CredentialSlot/CredentialSlotSerializer/mask_token/get_standardized_response_schemaqy_lty/aiapp/views.py文件末尾含class CredentialSlotAdminView(APIView):，含 GET/PUT 两个方法及各自@swagger_auto_schema装饰器qy_lty/userapp/admin_urls.py含path('credential-slot/', CredentialSlotAdminView.as_view(), name='admin_credential_slot')python manage.py check通过reverse('admin_credential_slot')解析到/api/v1/admin/credential-slot/- grep
aiapp/views.py内mask_token(instance.access_token)≥ 1 次（脱敏调用点） - grep
aiapp/views.py内if not request.user.is_staff≥ 1 次（admin 二次校验） - grep
aiapp/views.py不含RetrieveUpdateAPIView（确认未走通用 view） </success_criteria>