"""登录凭据(密码库)CRUD 路由 统一入口 /api/credentials,写操作(POST/PUT/DELETE)受 API Key 保护 (Tailscale 内网/本机放行,外部来源必须携带 X-API-Key)。 - 明文密码经 /{id}/password 专用接口解密返回(与账号 reveal 语义一致) - 2FA 动态码由服务端生成(secret 永不出库),/{id}/otp 响应 no-store 防缓存残留 - 录入 secret 与校验它解耦:拿不到当前码可先存(标记未验证),后续 /{id}/otp/verify 补验 """ from typing import List, Optional from fastapi import APIRouter, Depends, Query, Response from sqlmodel import Session from app.core.security import require_api_key from app.database import get_session from app.schemas.credential import ( CredentialCreate, CredentialRead, CredentialUpdate, OtpBindRequest, OtpVerifyRequest, ) from app.services import credential_service router = APIRouter(prefix="/api/credentials", tags=["credentials"]) @router.get("", response_model=List[CredentialRead], summary="凭据列表(密码库,支持搜索/过滤)") def list_credentials( q: Optional[str] = Query(default=None, description="搜索站点/用户名/备注"), login_type: Optional[str] = Query(default=None, description="登录方式:password/oauth/other"), has_otp: Optional[bool] = Query(default=None, description="是否已绑定 2FA"), session: Session = Depends(get_session), ): return credential_service.list_credentials( session, q=q, login_type=login_type, has_otp=has_otp ) @router.post( "", response_model=CredentialRead, status_code=201, summary="创建凭据(可同时录入 2FA,动态码可选校验)", dependencies=[Depends(require_api_key)], ) def create_credential(data: CredentialCreate, session: Session = Depends(get_session)): return credential_service.create_credential(session, data) @router.put( "/{credential_id}", response_model=CredentialRead, summary="更新凭据(密码:留空不改,空串清除)", dependencies=[Depends(require_api_key)], ) def update_credential( credential_id: int, data: CredentialUpdate, session: Session = Depends(get_session) ): return credential_service.update_credential(session, credential_id, data) @router.delete( "/{credential_id}", summary="删除凭据(账号/资产不受影响,仅解除关联)", dependencies=[Depends(require_api_key)], ) def delete_credential(credential_id: int, session: Session = Depends(get_session)) -> dict: return credential_service.delete_credential(session, credential_id) @router.get( "/{credential_id}/password", summary="查看凭据密码明文(解密返回)", dependencies=[Depends(require_api_key)], ) def reveal_password( credential_id: int, response: Response, session: Session = Depends(get_session) ) -> dict: # 明文密码禁止缓存(与 /otp 保持一致),避免浏览器/代理缓存或历史残留 response.headers["Cache-Control"] = "no-store" return credential_service.reveal_password(session, credential_id) @router.get( "/{credential_id}/otp", summary="获取当前 2FA 动态码(服务端生成,secret 不出库)", dependencies=[Depends(require_api_key)], ) def current_otp( credential_id: int, response: Response, session: Session = Depends(get_session) ) -> dict: # 动态码禁止缓存(浏览器/代理),避免有效期内残留 response.headers["Cache-Control"] = "no-store" return credential_service.current_otp(session, credential_id) @router.put( "/{credential_id}/otp", response_model=CredentialRead, summary="绑定/更换 2FA(动态码可选:提供则校验通过才落库,留空存为未验证)", dependencies=[Depends(require_api_key)], ) def bind_otp( credential_id: int, data: OtpBindRequest, session: Session = Depends(get_session) ): return credential_service.bind_otp(session, credential_id, data) @router.post( "/{credential_id}/otp/verify", response_model=CredentialRead, summary="补验证 2FA(提交当前动态码,通过后解除「未验证」标记)", dependencies=[Depends(require_api_key)], ) def verify_otp( credential_id: int, data: OtpVerifyRequest, session: Session = Depends(get_session) ): return credential_service.verify_otp(session, credential_id, data) @router.delete( "/{credential_id}/otp", response_model=CredentialRead, summary="解绑 2FA", dependencies=[Depends(require_api_key)], ) def unbind_otp(credential_id: int, session: Session = Depends(get_session)): return credential_service.unbind_otp(session, credential_id)