import json import math import time from typing import Any from uuid import UUID from fastapi import APIRouter, Depends, File, Header, Path, Query, Request, UploadFile # NB: this router intentionally returns raw JSONResponse for the Feishu # webhook handler. Feishu's push protocol MUST receive one of: # { "challenge": "" } — URL verification # { "code": 0 } — acknowledged # { "code": , "msg": "..." } — rejected (signature / token) # Wrapping these in Helix's standard ``{success,data,meta,errors}`` # envelope breaks Feishu callback verification and silently drops every # approval event. Do NOT refactor. from fastapi.responses import JSONResponse from pydantic import BaseModel from sqlalchemy.ext.asyncio import AsyncSession from app.approval.schemas import ( AdminBuybackOnBehalfRequest, AssetApprovalSubmitRequest, StageNoteUpdateRequest, ) from app.approval.service import ApprovalService from app.audit.service import AuditService from app.auth.models import AuthUser from app.core.database import get_db from app.core.dependencies import ( check_webhook_rate_limit, get_current_user, get_real_ip, require_admin, require_admin_read, require_asset_read, ) from app.core.exceptions import AppException from app.core.rbac import can_manage_assets from app.core.response import paginated_response, success_response # NB: this router intentionally returns raw JSONResponse for the Feishu # webhook handler. Feishu's push protocol MUST receive one of: # { "challenge": "" } — URL verification # { "code": 0 } — acknowledged # { "code": , "msg": "..." } — rejected (signature / token) # Wrapping these in Helix's standard ``{success,data,meta,errors}`` # envelope breaks Feishu callback verification and silently drops every # approval event. Do NOT refactor. from app.core.safe_logging import get_safe_logger from app.core.upload import read_upload_bounded from app.integrations.compat import get_feishu_config logger = get_safe_logger(__name__) # Widget-type-aware MIME allowlists (2026-05-19 split): # # Feishu has TWO separate upload endpoints with DIFFERENT format # restrictions, and ``upload_feishu_file`` dispatches between them by # widget_type — so a single allowlist either: # - over-rejects attachments (PDF rejected on the image widget makes # sense; rejecting it on attachmentV2 is wrong), OR # - over-accepts images (HEIC passes the Helix gate, then Feishu's # ``/im/v1/images`` returns 400 with a non-localised error the # user can't act on). # # ``image`` widget → ``/im/v1/images`` for photos (image_type=message, # Feishu support: PNG / JPEG / GIF / BMP / WebP, up to 10 MB; HEIC / # HEIF / TIFF / SVG NOT accepted — iPhone users need to switch # 设置 → 相机 → 格式 → 兼容性最佳 (JPEG) before shooting, OR re-save as # JPEG in 照片 app) **AND** ``/im/v1/files`` for videos when the Feishu # admin has enabled the 视频 checkbox on the widget (the buyback / # asset-return ``恢复出厂设置证明`` flow needs this — the form copy # literally requires a factory-reset video). The router dispatches by # MIME and ``ApprovalService.upload_feishu_file`` routes to the # matching Feishu endpoint; the form serialiser then emits # ``image_key`` for photos and ``file_key`` for videos in the same # value array (Feishu accepts both in one widget when the checkbox is # set). Feishu's /im/v1/files cap for video is 30 MB. # # ``attachmentV2`` widget → ``/drive/v1/medias/upload_all``. Feishu's # documented support: any file type up to 30 MB. IMAGE_PHOTO_MIME_TYPES = { "image/jpeg", "image/jpg", "image/pjpeg", "image/png", "image/gif", "image/bmp", "image/webp", } IMAGE_VIDEO_MIME_TYPES = { "video/mp4", "video/quicktime", # iPhone .mov "video/x-m4v", "video/webm", "video/x-matroska", # .mkv "video/3gpp", "video/x-msvideo", # .avi } IMAGE_WIDGET_MIME_TYPES = IMAGE_PHOTO_MIME_TYPES | IMAGE_VIDEO_MIME_TYPES ATTACHMENT_WIDGET_MIME_TYPES = IMAGE_PHOTO_MIME_TYPES | IMAGE_VIDEO_MIME_TYPES | { "image/heic", "image/heif", "image/tiff", "application/pdf", "application/msword", "application/vnd.openxmlformats-officedocument.wordprocessingml.document", "application/vnd.ms-excel", "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", "application/vnd.ms-powerpoint", "application/vnd.openxmlformats-officedocument.presentationml.presentation", "text/plain", "text/csv", # Some clients (older Win Outlook, Linux without a registered # handler) send Office docs as octet-stream. Feishu's # /drive/v1/medias/upload_all accepts those; let them through. "application/octet-stream", "application/zip", } # All approval-form uploads go through Feishu's special endpoint # ``/approval/openapi/v2/file/upload``, whose documented ceilings are # 10 MB per image and 50 MB per file (videos ride the ``type=attachment`` # path, so they get the 50 MB ceiling too). 2026-06-11: a raw iPhone # 电脑初始化 video blew past the previous 30 MB cap and died at nginx # with a bare 413 (HTML body → frontend showed the useless axios # "Request failed with status code 413"); raised to Feishu's real # limit. The frontend ``web/nginx.conf`` ``client_max_body_size`` # carries headroom above this so the proxy never rejects before this # gate fires — keep the two in sync when changing either. MAX_PHOTO_UPLOAD_SIZE = 10 * 1024 * 1024 MAX_VIDEO_UPLOAD_SIZE = 50 * 1024 * 1024 MAX_ATTACHMENT_UPLOAD_SIZE = 50 * 1024 * 1024 from app.core.dependencies import ( # noqa: E402 require_active_employee_for_mutations as _leaver_runtime_gate, ) router = APIRouter( prefix="/api/v1/approval", tags=["approval"], dependencies=[Depends(_leaver_runtime_gate)], ) @router.get("/asset/templates") async def list_asset_templates( request: Request, current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): service = ApprovalService(db) templates = await service.list_asset_templates() return success_response(data=templates, request=request) @router.post("/asset/templates/refresh") async def refresh_asset_templates( request: Request, current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Admin-only: drop the cache and re-fetch templates from Feishu. Exposed for the "立即刷新" button on /admin pages. Returns the freshly-fetched template list so the caller can update its UI. """ if not can_manage_assets(current_user.role_names): raise AppException( code="NOT_AUTHORIZED", message="Only asset/IT admins can force-refresh approval templates.", status_code=403, ) service = ApprovalService(db) await service.invalidate_template_cache() templates = await service.list_asset_templates(force_refresh=True) audit = AuditService(db) await audit.log( trace_id=request.state.trace_id, actor_type="user", actor_user_id=current_user.id, action="approval.template_cache.refresh", status="success", resource_type="platform_setting", resource_id="approval.asset_template_cache", detail={"count": len(templates)}, ) await db.commit() return success_response(data=templates, request=request) @router.get("/admin/flow-stages") async def list_admin_flow_stages( request: Request, refresh: bool = Query(False, description="Force re-fetch from Feishu"), current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """List every Feishu-backed approval flow's stage chain + admin notes. Backs the IT-admin "流程节点备注" Modal where IT can review every flow (asset / software / cloud / security) and curate per-stage hints visible to employees on the create-request preview. Admin-only — same gate as the asset-template refresh endpoint (``can_manage_assets``). ``refresh=true`` busts the cache and re-fetches all definitions from Feishu in parallel. """ if not can_manage_assets(current_user.role_names): raise AppException( code="NOT_AUTHORIZED", message="Only asset/IT admins can manage approval stage notes.", status_code=403, ) service = ApprovalService(db) if refresh: await service.invalidate_flow_stages_cache() bundles = await service.list_all_flow_stages(force_refresh=refresh) if refresh: audit = AuditService(db) await audit.log( trace_id=request.state.trace_id, actor_type="user", actor_user_id=current_user.id, action="approval.flow_stages_cache.refresh", status="success", resource_type="platform_setting", resource_id="approval.flow_stages_cache", detail={"count": len(bundles)}, ) await db.commit() return success_response(data=bundles, request=request) @router.put("/admin/stage-notes") async def upsert_admin_stage_note( request: Request, body: StageNoteUpdateRequest, current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Upsert one admin note keyed by (approval_code, node_id). ``note=null`` (or empty string) deletes the entry. Audit logged. """ if not can_manage_assets(current_user.role_names): raise AppException( code="NOT_AUTHORIZED", message="Only asset/IT admins can edit approval stage notes.", status_code=403, ) service = ApprovalService(db) await service.set_stage_note( approval_code=body.approval_code, node_id=body.node_id, note=body.note, ) audit = AuditService(db) await audit.log( trace_id=request.state.trace_id, actor_type="user", actor_user_id=current_user.id, action="approval.stage_note.upsert" if body.note else "approval.stage_note.delete", status="success", resource_type="platform_setting", resource_id=f"approval.stage_notes:{body.approval_code}:{body.node_id}", detail={ "approval_code": body.approval_code, "node_id": body.node_id, "note_length": len(body.note) if body.note else 0, }, ) await db.commit() return success_response(data={"ok": True}, request=request) @router.get("/flow-stages/{approval_code}") async def get_flow_stages_by_code( approval_code: str = Path(min_length=1, max_length=128), request: Request = None, # type: ignore[assignment] current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Public read of ONE flow's stages + notes. Used by non-asset create modals (api_key / vpn / dns / sso / ...) so they can render the same ApprovalStagesPreview the asset templates do. Any authenticated user — the notes are designed for employee consumption. """ service = ApprovalService(db) bundle = await service.get_flow_stages_by_code(approval_code) if bundle is None: raise AppException( code="FLOW_NOT_FOUND", message=f"No approval flow on file for code {approval_code}", status_code=404, ) return success_response(data=bundle, request=request) @router.get("/flow-stages-by-key/{flow_key}") async def get_flow_stages_by_key( flow_key: str = Path(min_length=1, max_length=64), request: Request = None, # type: ignore[assignment] current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Public read of ONE flow's bundle resolved by ``flow_key``. Lets each non-asset Modal hit one URL with its hardcoded ``flow_key`` constant (``api_key_request`` / ``vpn_application_request`` / ``google_mail_group`` / ...) without knowing the per-tenant ``approval_code``. Same response shape as ``/flow-stages/{approval_code}`` so the frontend wrapper is polymorphic on either lookup mode. """ service = ApprovalService(db) bundle = await service.get_flow_stages_by_key(flow_key) if bundle is None: raise AppException( code="FLOW_NOT_FOUND", message=f"No approval flow on file for flow_key {flow_key}", status_code=404, ) return success_response(data=bundle, request=request) @router.get("/asset/requests") async def list_asset_approval_requests( request: Request, page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100), status: str | None = Query(None), scope: str | None = Query(None, description="Set to 'mine' to force owner-scope even for admins"), request_type: list[str] | None = Query(None), keyword: str | None = Query(None), requester_id: UUID | None = Query( None, description="Filter to a single applicant (admin views)", ), start_date: str | None = Query(None), end_date: str | None = Query(None), current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): service = ApprovalService(db) requests, total = await service.list_asset_approval_requests( current_user=current_user, page=page, page_size=page_size, status=status, scope=scope, request_type=request_type, keyword=keyword, requester_id=requester_id, start_date=start_date, end_date=end_date, ) return paginated_response( data=requests, total=total, page=page, page_size=page_size, request=request, ) @router.get("/asset/requests/summary") async def asset_requests_summary( request: Request, scope: str | None = Query(None, description="Set to 'mine' to force owner-scope even for admins"), request_type: list[str] | None = Query(None), window_days: int = Query(30, ge=1, le=180), current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Scoped aggregate for the employee-request page charts. Same scope semantics as ``GET /asset/requests`` (non-admins see only their own; admins see all, narrowable by request_type) so the chart population matches the list. Returns by_status / by_request_type distributions + a windowed daily-created trend. Read-only. """ service = ApprovalService(db) data = await service.asset_request_summary( current_user=current_user, scope=scope, request_type=request_type, window_days=window_days, ) return success_response(data=data, request=request) @router.get("/admin/analytics") async def get_admin_approval_analytics( request: Request, window_days: int = 30, _user: AuthUser = Depends(require_admin_read()), db: AsyncSession = Depends(get_db), ): """Approval-center aggregate signal for the 分析 tab. Read-only; surfaces volume, status pie, by-type breakdown, turnaround stats, daily timeseries, top requesters. The window clamps to 1..180d — approval volume is per-employee bounded so 180d is well under a million rows even at our growth ceiling. """ # ``is None`` check, not falsy short-circuit: an explicit # ``?window_days=0`` should clamp to 1, not silently resolve to # the default 30 via ``0 or 30``. window_days = max(1, min(int(window_days if window_days is not None else 30), 180)) from app.approval.admin_overview import build_admin_analytics_summary data = await build_admin_analytics_summary(db, window_days=window_days) return success_response(data=data, request=request) @router.get("/admin/overview") async def get_admin_approval_overview( request: Request, source: str | None = Query(None), decided_in: str | None = Query(None), q: str | None = Query(None, max_length=100), page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100), _user: AuthUser = Depends(require_admin_read()), db: AsyncSession = Depends(get_db), ): """Platform-wide pending-approval board for the 待办总览 tab. Read-only. Aggregates every approval-type item awaiting a decision — employee_requests, cloud scope-requests, leaver dual-sign overrides, DNS / domain / VPN / AI-tool / announcement requests, API-key releases — into one oldest-pending-first list, each tagged with where the decision is made (helix / feishu) and a deep-link to that surface. Unifies discovery; decisions still happen on the owning page. """ from app.approval.admin_overview import build_admin_approval_overview data = await build_admin_approval_overview( db, source=source, decided_in=decided_in, q=q, page=page, page_size=page_size, ) return success_response(data=data.model_dump(), request=request) @router.get("/admin/overview/export.xlsx") async def export_admin_approval_overview( request: Request, source: str | None = Query(None), decided_in: str | None = Query(None), q: str | None = Query(None, max_length=100), current_user: AuthUser = Depends(require_admin_read()), db: AsyncSession = Depends(get_db), ): """Export the filtered 待办总览 board as XLSX. Filter signature mirrors :func:`get_admin_approval_overview` minus pagination; RBAC mirrors it too (``require_admin_read``). Reuses ``build_admin_approval_overview`` with ``page=1, page_size=`` so every provider / filter fix on the list applies to the export automatically. Capped at :data:`app.core.exporting.EXPORT_ROW_LIMIT` rows. """ from app.approval.admin_overview import build_admin_approval_overview from app.core.exporting import EXPORT_ROW_LIMIT, xlsx_export_response data = await build_admin_approval_overview( db, source=source, decided_in=decided_in, q=q, page=1, page_size=EXPORT_ROW_LIMIT, ) columns = [ ("source", "来源"), ("decided_in", "决策面"), ("requester_display", "申请人"), ("requester_email", "申请人邮箱"), ("requester_employee_no", "申请人工号"), ("title", "标题"), ("subtitle", "副标题"), ("created_at", "提交时间"), ("age_days", "等待天数"), ("id", "ID"), ] rows = [item.model_dump() for item in data.items] return await xlsx_export_response( db, request, current_user, columns=columns, rows=rows, total_matched=data.total, filename_stem="approval_overview", sheet_title="Approval Overview", audit_action="approval.admin_overview.export", audit_resource_type="approval_overview", filters={"source": source, "decided_in": decided_in, "q": q}, ) @router.get("/admin/all-requests") async def list_admin_all_requests( request: Request, source: str | None = Query(None), request_type: str | None = Query(None), status: str | None = Query(None), requester_id: UUID | None = Query(None), q: str | None = Query(None, max_length=100), start_date: str | None = Query(None), end_date: str | None = Query(None), page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100), _user: AuthUser = Depends(require_admin_read()), db: AsyncSession = Depends(get_db), ): """Platform-wide ALL-STATUS request list for the 申请列表 tab. Read-only. The cross-domain sibling of ``/admin/overview``: every approval-type request in any lifecycle state (pending / approved / rejected / completed / failed / cancelled), across all sources — employee_requests (asset / software / api-key / google), cloud scope-requests, DNS / domain / VPN / AI-tool / announcement requests, API-key releases, leaver dual-sign overrides, identity-consumer requests — in one newest-first table. ``source`` narrows to one domain; ``status`` is the canonical lifecycle filter; ``request_type`` further narrows the asset/software source by raw sub-type. This is the lifecycle "闭环": browse and search the whole request history, not just one table. Decisions still happen on each row's owning page (deep_link) or in Feishu. """ from app.approval.admin_overview import build_admin_request_list data = await build_admin_request_list( db, source=source, request_type=request_type, status=status, requester_id=requester_id, q=q, start_date=start_date, end_date=end_date, page=page, page_size=page_size, ) return success_response(data=data.model_dump(), request=request) @router.get("/admin/sources") async def list_admin_approval_source_catalog( request: Request, _user: AuthUser = Depends(require_admin_read()), ): """Server-owned approval source catalogue for filters and capabilities.""" from app.approval.admin_overview import list_admin_approval_sources return success_response( data=[item.model_dump() for item in list_admin_approval_sources()], request=request, ) @router.get("/admin/all-requests/detail") async def get_admin_unified_request_detail( request: Request, id: str = Query(..., max_length=160), # noqa: A002 - public API contract _user: AuthUser = Depends(require_admin_read()), db: AsyncSession = Depends(get_db), ): """Read one cross-domain approval detail without changing its workflow. The source-owned row remains authoritative. Live Feishu timeline lookup is read-only and degrades to the persisted lifecycle when Feishu is unavailable. """ from app.domains.me_universe.request_detail import build_my_request_detail data = await build_my_request_detail( db, person_id=None, item_id=id, enforce_owner=False, with_live_approval=True, ) return success_response(data=data.model_dump(), request=request) @router.get("/admin/all-requests/export.xlsx") async def export_admin_all_requests( request: Request, source: str | None = Query(None), request_type: str | None = Query(None), status: str | None = Query(None), requester_id: UUID | None = Query(None), q: str | None = Query(None, max_length=100), start_date: str | None = Query(None), end_date: str | None = Query(None), current_user: AuthUser = Depends(require_admin_read()), db: AsyncSession = Depends(get_db), ): """Export the filtered 申请列表 as XLSX. Filter signature mirrors :func:`list_admin_all_requests` minus pagination; RBAC mirrors it (``require_admin_read``). Reuses ``build_admin_request_list`` with ``page=1, page_size=`` so every source / filter fix applies to the export automatically. Capped at :data:`app.core.exporting.EXPORT_ROW_LIMIT` rows. """ from app.approval.admin_overview import build_admin_request_list from app.core.exporting import EXPORT_ROW_LIMIT, xlsx_export_response data = await build_admin_request_list( db, source=source, request_type=request_type, status=status, requester_id=requester_id, q=q, start_date=start_date, end_date=end_date, page=1, page_size=EXPORT_ROW_LIMIT, ) columns = [ ("source", "来源"), ("decided_in", "决策面"), ("requester_display", "申请人"), ("requester_email", "申请人邮箱"), ("requester_employee_no", "申请人工号"), ("title", "标题"), ("subtitle", "副标题"), ("status_normalized", "状态"), ("status", "原始状态"), ("created_at", "提交时间"), ("processed_at", "处理时间"), ("id", "ID"), ] rows = [item.model_dump() for item in data.items] return await xlsx_export_response( db, request, current_user, columns=columns, rows=rows, total_matched=data.total, filename_stem="approval_requests", sheet_title="Approval Requests", audit_action="approval.admin_all_requests.export", audit_resource_type="approval_requests", filters={ "source": source, "request_type": request_type, "status": status, "q": q, }, ) class _AdminUrgeBody(BaseModel): id: str # "{source}:{pk}" tracker item id @router.post("/admin/overview/urge") async def urge_admin_overview_item( request: Request, body: _AdminUrgeBody, current_user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ): """Admin 催办 from the 待办总览 board — nudge the approver(s) of any still-open request. Gated on ``require_admin`` (global_admin only) to MATCH the page's route guard (``RoleRoute roles={['global_admin']}`` on ``/admin/process-approval``). The cross-domain board lets one urge any request regardless of category, so a narrower role (asset_admin / auditor / management) hitting this endpoint directly would be acting beyond its authority on rows it can't even see — global_admin is the only role with org-wide reach. Rate-limited to once per request per 24h (shared with the applicant urge). Returns ``urged=False`` + i18n ``reason`` when terminal / unsupported / already urged. """ from app.domains.me_universe.urge import urge_request result = await urge_request( db, item_id=body.id, actor=current_user, is_admin=True, ) return success_response(data=result.model_dump(), request=request) @router.get("/admin/requests") async def list_admin_approval_requests( request: Request, page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100), status: str | None = Query(None), request_type: list[str] | None = Query(None), requester_id: UUID | None = Query(None), keyword: str | None = Query(None), start_date: str | None = Query(None), end_date: str | None = Query(None), _user: AuthUser = Depends(require_admin_read()), db: AsyncSession = Depends(get_db), ): """Unified admin view of every employee_requests row. IT-admin only. Backs the ``/admin/approvals`` page — ALL request kinds (asset hardware, software seat, api-key, google_*) in one paginated table with keyword search across request_number / reason / requester name / requester email. """ service = ApprovalService(db) items, total = await service.list_admin_approval_requests( page=page, page_size=page_size, status=status, request_type=request_type, requester_id=requester_id, keyword=keyword, start_date=start_date, end_date=end_date, ) return paginated_response( data=items, total=total, page=page, page_size=page_size, request=request, ) @router.get("/admin/requests/{request_id}") async def get_admin_approval_request_detail( request: Request, request_id: UUID = Path(...), refresh: bool = Query( True, description=( "If true (default), pull a live ``/approval/v4/instances/{code}``" " from Feishu so the timeline reflects the latest approver" " action. Set to false to read the cached snapshot only." ), ), _user: AuthUser = Depends(require_asset_read()), db: AsyncSession = Depends(get_db), ): """Return one employee_request enriched with the live Feishu timeline. Auth gate is ``require_asset_read`` (global_admin + asset_admin + auditor + management — read-only governance view), NOT the stricter ``require_admin`` (global_admin only). Reason: the admin EmployeeRequestPage at ``/admin/assets/{employee,software}-requests`` is reachable by ``asset_admin`` via the frontend route guard (``web/src/App.tsx`` line ~405 admits ``[global_admin, asset_admin]``). The same role gates the asset-domain LIST endpoint (``/asset/requests`` returns all rows for any admin via ``can_manage_assets``), so an asset_admin can already see every asset-request id in their list — refusing detail access for those same ids was a 2026-05-15 regression introduced by P0547's fix (admin-view now correctly calls this endpoint instead of the owner-scoped ``/my/requests/{id}``; previously the 404 from the owner-scope check accidentally masked this gap as a "Could not refresh from Feishu" warning while still rendering the cached list-row data). The all-approvals unified LIST at ``/admin/requests`` stays strict (``require_admin``) because that page is global-admin-only by frontend route + comment at ``App.tsx:266``. Drift reconcile (2026-05-28): when ``refresh=True`` AND the caller has write authority (``can_manage_assets`` — global_admin / asset_admin) AND the live Feishu fetch shows a terminal state but Helix is still pending, the missed callback is replayed inline so status + auto-execute side-effects catch up on the same click. Auditor + management see the refreshed timeline but never trigger a write — they remain a read-only governance surface. On Feishu fetch failure the response carries ``timeline_error`` plus whatever cached snapshot was available — the rest of the request data is still useful and we don't want a Feishu blip to 503 the entire admin page. """ service = ApprovalService(db) allow_reconcile = can_manage_assets(_user.role_names) detail = await service.get_admin_approval_request_detail( request_id=request_id, refresh_timeline=refresh, actor_user_id=_user.id, allow_reconcile=allow_reconcile, ) # The service may have persisted a refreshed Feishu snapshot back # into ``employee_requests.detail.feishu.latest_instance_detail`` # via ``flush()``; commit so subsequent admins (and the next # callback) read the same data. await db.commit() return success_response(data=detail, request=request) @router.get("/my/requests/{request_id}") async def get_my_approval_request_detail( request: Request, request_id: UUID = Path(...), refresh: bool = Query( True, description=( "If true (default), pull a live " "``/approval/v4/instances/{code}`` from Feishu so the" " timeline + progress reflect the latest approver" " action. Set to false to read the cached snapshot only." ), ), current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Owner-scoped twin of ``/admin/requests/{id}``. Returns the same enriched detail (form-snapshot blob, timeline, progress) for a request the caller owns. Surfaces ``NotFound`` rather than 403 if the row belongs to someone else so the endpoint cannot be used as an existence oracle for other employees' request ids. Admins routinely use the admin endpoint instead — there's no harm in admins calling this one too, but the route exists primarily for the "我的申请" detail drawer so employees see the same approval-progress UX as IT. """ service = ApprovalService(db) # Owner-scoped endpoint: the requester themself can always # reconcile their own row (they triggered the refresh because their # status is stuck). Auditor / management never hit THIS endpoint — # they use ``/admin/requests/{id}`` with its own gating. So # ``allow_reconcile=True`` is safe here: the owner-scope check # inside the service guarantees the caller owns the row. detail = await service.get_admin_approval_request_detail( request_id=request_id, refresh_timeline=refresh, owner_user_id=current_user.id, actor_user_id=current_user.id, allow_reconcile=True, ) # Commit so any refreshed Feishu snapshot persisted by the # service is visible to subsequent callers — same rationale as # the admin endpoint. await db.commit() return success_response(data=detail, request=request) @router.get("/asset/buyback-quote") async def get_offboarding_buyback_quote( request: Request, asset_id: UUID = Query(..., description="Asset to quote"), currency: str | None = Query( None, description=( "Legacy display-currency fallback for assets without a registered " "company currency. A registered purchase entity is authoritative." ), ), current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Compute the offboarding-buyback amount for an asset. Surface for the employee-facing 离职回购 form: when the requester picks an asset, the frontend calls this endpoint and renders the returned ``amount`` read-only (employees can't haggle the price). The currency selector is editable; switching it issues a fresh fetch with ``currency=...`` so the response always reflects the policy-FX-converted figure for that currency. """ service = ApprovalService(db) quote = await service.quote_offboarding_buyback( asset_id=asset_id, current_user=current_user, target_currency=currency, ) return success_response(data=quote, request=request) @router.get("/asset/payment-info") async def get_asset_payment_info( request: Request, asset_id: UUID = Query(..., description="Asset to fetch payment info for"), current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Return formatted payment info for the asset's ``purchase_entity``. Powers the laptop_replacement + offboarding_buyback "Payment Information" widget — when the applicant picks an asset, the frontend fetches this so the bank-info field renders read-only + pre-filled with the entity's ``bank_name`` / ``bank_account`` / ``bank_account_name`` / etc. Returns ``text=""`` (200, not 404) when the asset has no ``purchase_entity`` or the named entity isn't on file — the field stays editable in that case so the IT admin can fill it manually downstream. Authz mirrors the buyback quote: caller must own / currently borrow the asset, unless they're an asset/IT admin. Without this an employee could enumerate every asset's purchase_entity bank details by scripting against the endpoint. """ service = ApprovalService(db) asset = await service._get_asset(asset_id) if not can_manage_assets(current_user.role_names): if ( asset.assigned_to != current_user.id and asset.borrowed_by != current_user.id ): raise AppException( code="FORBIDDEN", message=( "You can only view payment info for assets you own " "or currently borrow." ), status_code=403, ) entity = await service._resolve_payment_entity(asset) return success_response( data={ "asset_id": str(asset.id), "purchase_entity": asset.purchase_entity, # Canonical resolved legal-entity name (``asset_legal_entities # .name``) — distinct from the free-text ``purchase_entity``. # The frontend matches this against the 所属主体 single-select # option set to pre-fill that widget. ``None`` when unresolved. "entity_name": entity.name if entity is not None else None, "has_entity": entity is not None, "text": ( ApprovalService._format_payment_info(entity) if entity is not None else "" ), }, request=request, ) @router.post("/asset/requests") async def create_asset_approval_request( request: Request, body: AssetApprovalSubmitRequest, current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), idempotency_key: str = Header(..., alias="Idempotency-Key"), ): from app.approval.submission_idempotency import ( SPECIALIZED_APPROVAL_TEMPLATE_KEYS, ) if body.template_key in SPECIALIZED_APPROVAL_TEMPLATE_KEYS: raise AppException( code="APPROVAL_SPECIALIZED_ENDPOINT_REQUIRED", message=( "Software seat and add-on approvals must use their specialized " "asset request endpoint." ), status_code=422, ) service = ApprovalService(db) response = await service.create_asset_approval_request( body=body, current_user=current_user, trace_id=request.state.trace_id, ip_address=get_real_ip(request), user_agent=request.headers.get("User-Agent"), idempotency_key=idempotency_key, ) await db.commit() return success_response(data=response, request=request) @router.post("/asset/requests/on-behalf") async def create_buyback_request_on_behalf( request: Request, body: AdminBuybackOnBehalfRequest, current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), idempotency_key: str = Header(..., alias="Idempotency-Key"), ): """Admin 代提回购 — IT starts the offboarding-buyback approval for a leaver-held asset on the departed employee's behalf. Role gate is enforced inside the service (``can_manage_assets``), which also resolves the buyback subject from the asset's current holder and rejects an asset with no holder. The admin becomes the Feishu applicant because the leaver's Feishu account can't submit; the stored requester is the leaver so the approve-time disposal is attributed correctly. """ service = ApprovalService(db) response = await service.create_offboarding_buyback_on_behalf( asset_id=body.asset_id, current_user=current_user, trace_id=request.state.trace_id, ip_address=get_real_ip(request), user_agent=request.headers.get("User-Agent"), reason=body.reason, notes=body.notes, currency=body.currency, base_location=body.base_location, last_working_day=body.last_working_day, idempotency_key=idempotency_key, ) await db.commit() return success_response(data=response, request=request) @router.get("/asset/requests/on-behalf/context") async def get_buyback_request_on_behalf_context( request: Request, asset_id: UUID = Query(...), current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Return admin-only HR context needed to render a safe buyback form.""" service = ApprovalService(db) response = await service.get_offboarding_buyback_on_behalf_context( asset_id=asset_id, current_user=current_user, ) return success_response(data=response, request=request) @router.post("/asset/uploads") async def upload_asset_approval_file( request: Request, widget_type: str = Query(...), flow: str = Query( "form", description=( "'supplement' = the file is destined for a post-submit " "supplement comment; a durable copy is persisted to Helix " "S3 because comment attachments have no re-sign channel " "(the Feishu CDN url expires in ~14h and the comment is " "never re-rendered with a fresh signature, unlike form " "widget files which re-sign on every instance fetch). " "'form' = create-flow widget upload, no copy needed." ), ), file: UploadFile = File(...), current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): # Widget-type-aware MIME validation (2026-05-19). The two Feishu # upload endpoints (/im/v1/images vs /drive/v1/medias/upload_all) # have different supported formats — a single allowlist either # over-rejects attachments OR over-accepts images and lets Feishu's # opaque 230002 surface. Use ``AppException`` (not bare # ``HTTPException``) so the standard ``{success,data,errors}`` # envelope handler fires and the frontend's ``getApiErrorMessage`` # reads ``errors[0].message``. content_type = (file.content_type or "").lower() if widget_type == "image": allowed = IMAGE_WIDGET_MIME_TYPES hint = ( "图片/视频字段支持 JPG / PNG / GIF / BMP / WebP 图片，或 MP4 / " "MOV / WebM 视频。iPhone HEIC 请在「设置 → 相机 → 格式 → " "兼容性最佳」改成 JPEG 后重拍。" ) max_size = ( MAX_VIDEO_UPLOAD_SIZE if content_type.startswith("video/") else MAX_PHOTO_UPLOAD_SIZE ) elif widget_type == "attachmentV2": allowed = ATTACHMENT_WIDGET_MIME_TYPES hint = "附件字段支持图片 / 视频 / PDF / Word / Excel / PPT 等常见文件。" max_size = MAX_ATTACHMENT_UPLOAD_SIZE else: await file.close() raise AppException( code="UNSUPPORTED_UPLOAD_WIDGET", message=f"Unsupported upload widget type: {widget_type}", status_code=400, ) if content_type not in allowed: await file.close() raise AppException( code="UPLOAD_TYPE_NOT_ALLOWED", message=( f"不支持的文件类型 ({file.content_type or 'unknown'})。{hint}" ), status_code=400, ) content_length = request.headers.get("content-length") if content_length and int(content_length) > max_size + 1024 * 1024: # Cheap early reject on the header (multipart envelope adds # ~KBs, allow 1 MB slack); the exact gate below re-checks the # actual file bytes. await file.close() raise AppException( code="UPLOAD_TOO_LARGE", message=f"文件过大，单个文件最多 {max_size // (1024 * 1024)}MB。", status_code=413, ) file_bytes = await read_upload_bounded( file, max_size, message=f"文件过大，单个文件最多 {max_size // (1024 * 1024)}MB。", close=True, ) from app.core.storage_uploads import enforce_approval_upload_quota await enforce_approval_upload_quota(current_user.id, len(file_bytes)) if file.filename: # Stricter sanitiser — the previous ``replace("/", "_")`` / # ``replace("..", "_")`` chain didn't strip control bytes # (``\x00``, ``\r``, ``\n``), URL-encoded traversal # (``%2e%2e%2f``), or unicode-lookalike dots, so the cleaned # filename could still render as a traversal hint in admin UI # or carry confusable glyphs into the audit detail. Positive # regex + length cap mirrors ``storage_router._safe_filename``. from app.storage_router import _safe_filename as _safe_fn # type: ignore[attr-defined] try: file.filename = _safe_fn(file.filename) except Exception: # noqa: BLE001 — fall back to the local helper import re as _re no_path = file.filename.replace("\\", "/").rsplit("/", 1)[-1] cleaned = _re.sub(r"[^A-Za-z0-9._\-]", "_", no_path).strip("._") or "file" file.filename = cleaned[-120:] # Supplement files ALSO get a durable Helix S3 copy. The Feishu # comment attachment renders from a signed CDN url that expires in # ~14h with no re-sign channel (comments are stored verbatim, # unlike form widget values which re-sign on every instance GET) — # without this copy the supplement history in Helix goes # permanently dark once the signature lapses. Best-effort: a # storage outage must not block the Feishu upload (the approver's # hot-review window matters more), so failures log + degrade to # the ephemeral url. object_key: str | None = None if flow == "supplement": from datetime import UTC as _UTC from datetime import datetime as _dt from uuid import uuid4 as _uuid4 from app.core.storage import StorageNotConfiguredError, put_object_bytes try: today = _dt.now(_UTC).strftime("%Y/%m/%d") rel_key = ( f"approval_supplement/{today}/{current_user.id}/" f"{_uuid4().hex[:12]}-{file.filename or 'file'}" ) object_key = await put_object_bytes( db, relative_key=rel_key, content=file_bytes, content_type=content_type or "application/octet-stream", ) except StorageNotConfiguredError: logger.warning( "approval supplement upload: storage not configured — " "skipping durable copy for %s", file.filename, ) except Exception: # noqa: BLE001 — degrade, don't block Feishu path logger.exception( "approval supplement upload: S3 copy failed for %s — " "continuing with ephemeral Feishu url only", file.filename, ) service = ApprovalService(db) uploaded = await service.upload_feishu_file( widget_type, filename=file.filename or "file", content=file_bytes, content_type=content_type or "application/octet-stream", ) # All approval-form file uploads now go through Feishu's special # approval upload endpoint ``https://www.feishu.cn/approval/openapi/ # v2/file/upload``, which returns ``{"code": "", "url": "..."}``. # The ``code`` is the canonical file identifier the image / # attachmentV2 widget value arrays must reference (2026-05-26 # rabbit hole: previously we uploaded to ``/im/v1/images`` / # ``/im/v1/files`` / ``/drive/v1/medias/upload_all`` and got # ``img_v3_*`` / ``file_v2_*`` / ``file_token`` tokens — Feishu # ACCEPTED the form payload but rendered the widget as EMPTY in # the approval card because those tokens reference IM / Drive # files, not approval files). Legacy field-name fallbacks kept # so a Feishu version that returns the older shape on the same # endpoint still resolves. token = ( uploaded.get("code") or uploaded.get("file_token") or uploaded.get("file_key") or uploaded.get("image_key") ) # The approval upload endpoint also returns a viewable ``url`` # (Feishu drive stream link, opens for any logged-in Feishu user). # Pass it through — the supplement-uploads flow persists it so the # admin / requester can actually OPEN a supplemented video later # (the bare token is useless to a human; supplement files live in # the approval-instance comment, so they never appear in # ``latest_instance_detail.form`` where the other viewable URLs # come from). url = uploaded.get("url") return success_response( data={ "token": token, "widget_type": widget_type, "file_name": file.filename, "url": url if isinstance(url, str) and url else None, "object_key": object_key, "size": len(file_bytes), }, request=request, ) #: Feishu ``create_time`` values are milliseconds since epoch on the v2 #: envelope. A value this large cannot be a second-precision stamp (it would be #: the year 33658), so the unit is decided by magnitude rather than assumed. _MILLISECOND_EPOCH_FLOOR = 1_000_000_000_000 def _envelope_timestamp(header: Any) -> float | None: """Event time in epoch SECONDS from a v2 envelope header, or None. Exists because ``card.action.trigger`` carries no ``X-Lark-Request-*`` headers: the only freshness assertion available is the one inside the body that the verification token authenticates. """ if not isinstance(header, dict): return None raw = header.get("create_time") if raw is None or isinstance(raw, bool): return None try: value = float(str(raw).strip()) except (ValueError, TypeError): return None # ``isfinite`` FIRST, and it is the whole point of this line. ``nan <= 0`` is # False and ``nan >= 1e12`` is False, so a NaN sailed through both guards and # came back as the event time — and ``abs(now - nan) > 300`` is also False, so # the freshness check silently passed. A ``create_time`` of the three # characters "nan" therefore made a captured callback replayable forever, # which is precisely the bound the card-tap dedup exemption relies on. if not math.isfinite(value) or value <= 0: return None return value / 1000.0 if value >= _MILLISECOND_EPOCH_FLOOR else value @router.post("/feishu/callback") async def handle_feishu_callback( request: Request, db: AsyncSession = Depends(get_db), # A webhook-sized limiter, NOT the per-IP login limiter: all of Feishu's # org-wide approval callbacks (+ delivery retries) arrive from a small set # of shared egress IPs, so the 20/900s login ceiling tripped almost # immediately and fed a 429→Feishu-retry→429 loop that silently stalled # approval-state sync. HMAC/token verification + replay guards below are # the real auth; this only bounds a malformed-push flood. _rate: None = Depends(check_webhook_rate_limit), ): # Read raw body bytes so we can verify the HMAC signature before # parsing. Cap body size at 1MB so a malformed / oversized push # cannot trigger a heavy ``json.loads`` on a path that runs before # any auth check. body_bytes = await request.body() if len(body_bytes) > 1_000_000: return JSONResponse( {"code": 400, "msg": "payload too large"}, status_code=400, ) try: payload: dict[str, Any] = json.loads(body_bytes) except (ValueError, TypeError): return JSONResponse( {"code": 400, "msg": "invalid JSON"}, status_code=400, ) if not isinstance(payload, dict): return JSONResponse( {"code": 400, "msg": "payload must be a JSON object"}, status_code=400, ) service = ApprovalService(db) lark_sig = request.headers.get("X-Lark-Signature") lark_ts = request.headers.get("X-Lark-Request-Timestamp", "") lark_nonce = request.headers.get("X-Lark-Request-Nonce", "") feishu_cfg = await get_feishu_config(session=db) event_encrypt_key = str(feishu_cfg.get("event_encrypt_key") or "") header: dict[str, Any] = ( payload["header"] if isinstance(payload.get("header"), dict) else {} ) event_body: dict[str, Any] = ( payload["event"] if isinstance(payload.get("event"), dict) else {} ) # Freshness and replay identity as carried by the v2 envelope itself, for # the pushes that ship no ``X-Lark-Request-*`` headers at all (card # callbacks). ``create_time`` is milliseconds on every v2 push observed in # production; seconds are accepted too rather than trusting one unit — a # misread unit would put every event outside the window and fail closed on # a whole event type, which is the defect this replaces. envelope_ts = _envelope_timestamp(header) envelope_event_id = str(header.get("event_id") or "") if header else "" event_type_hint = ( payload.get("type") or header.get("event_type") or event_body.get("type") ) logger.info( "feishu callback received: has_sig=%s event_type=%s has_body_token=%s instance_code=%s", bool(lark_sig), event_type_hint, bool(payload.get("token") or header.get("token")), event_body.get("instance_code") or payload.get("instance_code"), ) # Unified verification. We accept the callback if ANY of the following # produce a match, in order: # 1. Valid X-Lark-Signature over the body (when both header and # event_encrypt_key are present) # 2. Verification Token matches at payload.token / header.token / # event.token # We reject when ALL of: no valid HMAC signature, AND verify_callback_token # raises (token mismatch, absent, or unconfigured — see service.verify_callback_token). sig_errors: list[str] = [] sig_ok = False if lark_sig and event_encrypt_key: try: ts_int = int(lark_ts) if abs(time.time() - ts_int) > 300: sig_errors.append(f"sig_ts_too_old ts={lark_ts}") elif service.verify_lark_signature( event_encrypt_key, lark_ts, lark_nonce, body_bytes, lark_sig ): sig_ok = True else: sig_errors.append("sig_mismatch") except (ValueError, TypeError): sig_errors.append(f"sig_bad_ts={lark_ts!r}") # Pre-launch P0 fix (Round 45, CR-101): SETNX nonce-replay guard on # the production HTTP endpoint. The driver's verify_webhook has the # same guard, but Feishu hits this router directly — so the driver # check is dead code on the live path. Without this, an attacker who # captures one valid HMAC-signed callback (or holds a leaked # verification token) can replay it within the 300s freshness # window: each replay re-fires service.handle_feishu_callback → # re-advances the approval state machine and re-runs # _auto_execute_approved_request (re-borrows / re-grants seats). # Fail-open on Redis fault so a Redis blip doesn't refuse # legitimate Feishu callbacks (replay risk during outage is # acceptable; permanent rejection is not). replay_err: str | None = None # The dedup nonce we successfully claimed (if any). Set BEFORE the # payload is processed, but processing (or its commit) can still fail; # if it does we must RELEASE the nonce so Feishu's redelivery is not # silently dropped as a replay, leaving the approval stuck ``pending`` # forever. Recorded here, deleted in the failure branch below. dedup_key_set: str | None = None if sig_ok and lark_ts and lark_nonce: try: from app.core.redis_client import get_redis # noqa: PLC0415 r = get_redis() if r is not None: key = f"approval:webhook_replay:{lark_nonce}:{lark_ts}" set_result = await r.set(key, "1", nx=True, ex=600) if not set_result: replay_err = ( "sig_path_replayed nonce+ts already processed " "within ±10min" ) else: dedup_key_set = key except Exception: # noqa: BLE001 — fail-open on Redis fault logger.warning( "feishu webhook replay cache unavailable on sig path — " "accepting without dedup", ) # Imported before the verification block: the replay-guard exemption below # asks this the same way the routing does. from app.assistant.card_actions import ( # noqa: PLC0415 claims_card_action, ) is_assistant_card_action = claims_card_action(payload) token_err: str | None = None is_url_verification = payload.get("type") == "url_verification" if not sig_ok: # Replay-protection on the token-only path. The signature path # already enforces a 5-minute freshness window above. Without a # matching guard here, a leaked verification token (it's a # static shared secret) can replay any past callback indefinitely # — re-approving / re-rejecting old approval instances and # re-firing seat activations. Require the same timestamp header # Feishu sends on every state-changing push. # # The ``url_verification`` handshake is exempt because (a) Feishu # sends it during integration setup before timestamp headers are # added, and (b) it's idempotent (re-replaying it just echoes # the challenge — no state change). if not is_url_verification: if lark_ts: try: ts_int = int(lark_ts) if abs(time.time() - ts_int) > 300: token_err = f"token_path_ts_too_old ts={lark_ts}" except (ValueError, TypeError): token_err = f"token_path_bad_ts={lark_ts!r}" elif envelope_ts is not None and envelope_event_id: # A ``card.action.trigger`` post carries NO # ``X-Lark-Request-Timestamp``, ``X-Lark-Request-Nonce`` or # ``X-Lark-Signature`` header — only the v2 envelope, whose # ``header`` holds ``create_time`` (ms) and ``event_id``. # Requiring the header therefore rejected every card tap with # 401, which the Feishu client renders as the opaque # 「Something went wrong … code: 200671」. Production audit, # 2026-08-18 16:17: three ``approval.callback.rejected`` rows # with ``token_reason=token_path_missing_timestamp``, one per # tap, and zero rows in ``assistant_message_feedback`` — every # button the assistant has ever sent was dead on arrival. # # ``create_time`` is the same assertion the header made, from # the signed-token-authenticated body instead of a header, so # the freshness window is unchanged; ``event_id`` replaces the # nonce for replay bounding below. # A card callback's ``create_time`` is the creation time of # the interactive message on some Feishu clients, rather than # the instant the employee taps it. Cards intentionally stay # usable in scrollback, so applying the five-minute transport # window here rejected perfectly valid feedback and escalation # taps before their idempotent handlers could run. These # actions are safe to replay (feedback is an upsert, escalation # is deduplicated, and proposal confirmation is conditional), # so use the event/action idempotency below instead. if ( not is_assistant_card_action and abs(time.time() - envelope_ts) > 300 ): token_err = f"token_path_envelope_ts_too_old ts={envelope_ts}" else: # No header timestamp, and no usable envelope either. Note the # ``event_id`` requirement above: without it the replay guard # below has nothing to key on, and accepting the push anyway # would mean a whole class of events with NO replay protection # at all. Feishu stamps every v2 push with one, so a push # missing it is not a client we need to support. token_err = "token_path_missing_timestamp" if token_err is None: try: await service.verify_callback_token(payload) except Exception as auth_exc: # noqa: BLE001 token_err = str(auth_exc) or type(auth_exc).__name__ # Pre-launch P0 fix (Round 45, CR-101): replay guard on the # token-only path. v1 callbacks don't ship X-Lark-Request-Nonce, # so key the dedup on the payload's instance_code / uuid plus # the timestamp — Feishu sends one callback per (instance, ts). if token_err is None and not is_url_verification and lark_ts: v1_dedup_id = ( payload.get("uuid") or event_body.get("instance_code") or payload.get("instance_code") ) if v1_dedup_id: try: from app.core.redis_client import get_redis # noqa: PLC0415 r = get_redis() if r is not None: key = f"approval:webhook_replay_v1:{v1_dedup_id}:{lark_ts}" set_result = await r.set( key, "1", nx=True, ex=600, ) if not set_result: token_err = ( "token_path_replayed instance+ts already " "processed within ±10min" ) else: dedup_key_set = key except Exception: # noqa: BLE001 — fail-open logger.warning( "feishu webhook replay cache unavailable on " "token path — accepting without dedup", ) # Same guard for the header-less v2 envelope, keyed on ``event_id``, # which Feishu mints once per event. # # A card tap is deliberately EXEMPT from the hard drop. Feishu re-posts # a callback it judged slow using the SAME ``event_id``, so no key can # tell its legitimate retry from a replay — and dropping the retry is # exactly the failure ``card_actions._already_done_card`` exists to # prevent (the employee reads 「没能提交」 for a submission that went # through). Every card branch is idempotent by construction instead: the # proposal confirm is a conditional ``pending → confirmed`` claim, the # verdict is an upsert on (message, person), and the escalation rides # the support service's own per-(requester, title) dedup window. The # 300-second freshness window above still bounds a replay. # # The exemption asks ``claims_card_action`` — the SAME predicate that # decides the routing below — rather than re-deriving the event type. # ``event_type_hint`` prefers the v1 top-level ``payload["type"]`` while # ``claims_card_action`` prefers ``header["event_type"]``, so a body # carrying different values in the two places could be exempted here and # then routed to the approval state machine anyway. Asking one question # once removes the disagreement rather than documenting it. if ( token_err is None and not is_url_verification and not lark_ts and envelope_ts is not None and envelope_event_id and not is_assistant_card_action ): try: from app.core.redis_client import get_redis # noqa: PLC0415 r = get_redis() if r is not None: key = f"approval:webhook_replay_v2:{envelope_event_id}" set_result = await r.set(key, "1", nx=True, ex=600) if not set_result: token_err = ( "token_path_replayed event_id already processed " "within 10min" ) else: dedup_key_set = key except Exception: # noqa: BLE001 — fail-open, as above logger.warning( "feishu webhook replay cache unavailable on envelope " "path — accepting without dedup", ) if replay_err: # Sig-path replay detection — log + drop. We don't need a 401 # because the original request was legitimate; the replay is a # quiet drop with audit row. audit = AuditService(db) await audit.log( trace_id=request.state.trace_id, actor_type="system", actor_user_id=None, action="approval.callback.replay_dropped", resource_type="approval", resource_id=None, status="failure", detail={ "reason": replay_err, "lark_ts": lark_ts, "lark_nonce": lark_nonce, }, ip_address=get_real_ip(request), user_agent=request.headers.get("User-Agent"), ) await db.commit() return JSONResponse({"code": 0, "msg": "replay dropped"}, status_code=200) if not sig_ok and token_err: # All verification paths failed — audit-log the rejection with a # payload snapshot so we can see exactly what Feishu sent. This is # the only way to diagnose Feishu push-shape drift without prod log # access. header_snapshot = payload.get("header") if isinstance(payload.get("header"), dict) else None event_snapshot = payload.get("event") if isinstance(payload.get("event"), dict) else None audit = AuditService(db) await audit.log( trace_id=request.state.trace_id, actor_type="system", actor_user_id=None, action="approval.callback.rejected", resource_type="approval", resource_id=None, status="failure", detail={ "token_reason": token_err, "sig_errors": sig_errors, "has_lark_sig": bool(lark_sig), "lark_ts": lark_ts, "lark_nonce": lark_nonce, "body_top_keys": list(payload.keys()), "header_keys": list(header_snapshot.keys()) if header_snapshot else None, "event_keys": list(event_snapshot.keys()) if event_snapshot else None, "header_event_type": header_snapshot.get("event_type") if header_snapshot else None, "header_app_id": header_snapshot.get("app_id") if header_snapshot else None, "event_app_id": event_snapshot.get("app_id") if event_snapshot else None, "top_level_app_id": payload.get("app_id"), "payload_type": payload.get("type"), "payload_token_present": payload.get("token") is not None, }, ip_address=get_real_ip(request), user_agent=request.headers.get("User-Agent"), ) await db.commit() return JSONResponse({"code": 401, "msg": "Invalid Feishu webhook token"}, status_code=401) if payload.get("type") == "url_verification": return JSONResponse({"challenge": payload.get("challenge")}) if payload.get("encrypt"): return JSONResponse( {"code": 1, "msg": "Encrypted Feishu callbacks are not configured in Helix yet"}, status_code=400, ) # A Feishu app has ONE event-subscription URL, so subscribing the assistant # to ``im.message.receive_v1`` delivers chat messages to this verified # handler too. Fork the event types the assistant owns; everything else # continues down the approval path untouched. The assistant acknowledges # immediately and answers in the background, because a model turn is far # slower than Feishu's redelivery window. # A card tap is answered INLINE, unlike a chat message: Feishu renders the # response body as the new card, so acknowledging first and working after # would leave the employee looking at a button that did nothing. from app.assistant.card_actions import handle_card_action # noqa: PLC0415 from app.assistant.feishu_events import ( # noqa: PLC0415 try_claim_event as try_claim_assistant_event, ) if is_assistant_card_action: # No session handed over: the handler owns its own, because a tap can # outlive this request (Feishu's 3s deadline vs a real domain flow). return JSONResponse( await handle_card_action(payload, trace_id=request.state.trace_id), ) if try_claim_assistant_event(payload, trace_id=request.state.trace_id): return JSONResponse({"code": 0, "msg": "success"}) try: await service.handle_feishu_callback(payload) event = payload.get("event") or {} instance_code = ( event.get("instance_code") or event.get("approval_instance_code") or payload.get("instance_code") ) audit = AuditService(db) await audit.log( trace_id=request.state.trace_id, actor_type="system", actor_user_id=None, action="approval.callback.processed", resource_type="approval", resource_id=instance_code, status="success", detail={ "event_type": payload.get("type") or payload.get("header", {}).get("event_type"), "approval_code": event.get("approval_code"), "instance_code": instance_code, "actor": "feishu_webhook", }, ip_address=get_real_ip(request), user_agent=request.headers.get("User-Agent"), ) await db.commit() except Exception: # Processing (or its commit) failed AFTER we claimed the replay # nonce. The nonce lives in Redis, outside this DB transaction, so # the rollback does NOT release it — and Feishu's automatic retry # of the SAME (nonce, ts) would be dropped as a "replay" and never # re-processed, stranding e.g. a REJECTED/CANCELLED callback with # ``status='pending'`` forever (the pending-reaper only rescues # APPROVED/COMPLETED). Release the nonce so the redelivery is # allowed to re-run, then propagate the 500 so Feishu retries. if dedup_key_set is not None: try: from app.core.redis_client import get_redis # noqa: PLC0415 r = get_redis() if r is not None: await r.delete(dedup_key_set) except Exception: # noqa: BLE001 — best-effort release logger.warning( "failed to release feishu webhook replay nonce %s after " "processing error; redelivery may be dropped as replay", dedup_key_set, ) raise return JSONResponse({"code": 0})