from __future__ import annotations import hashlib import hmac import io import json import uuid import zipfile from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any from urllib.parse import urlsplit, urlunsplit import httpx from fastapi import HTTPException, Request, status from jose import JWTError, jwt from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.core.config import settings from app.models.collaboration import ( CollaborationCallbackReceipt, CollaborationFile, CollaborationRevision, CollaborationSession, CollaborationShareLink, ) from app.models.user import User from app.schemas.collaboration import CollaborationCallbackPayload, CollaborationEditorConfigRead from app.services import collaboration_service, ledger_access, onlyoffice_service def collaboration_document_key(file_id: uuid.UUID, generation: int) -> str: fingerprint = f"{settings.ONLYOFFICE_INSTANCE_ID or ''}:collaboration:{file_id}:{generation}" return f"ctms-collab-{hashlib.sha256(fingerprint.encode('utf-8')).hexdigest()}" def _content_url(session_id: uuid.UUID) -> str: return ( f"{settings.ONLYOFFICE_STORAGE_BASE_URL.rstrip('/')}" f"/internal/onlyoffice/collaboration/sessions/{session_id}/content" ) def _callback_url(session_id: uuid.UUID) -> str: return ( f"{settings.ONLYOFFICE_STORAGE_BASE_URL.rstrip('/')}" f"/internal/onlyoffice/collaboration/sessions/{session_id}/callback" ) async def _active_session( db: AsyncSession, item: CollaborationFile, user_id: uuid.UUID ) -> CollaborationSession: session = await db.scalar( select(CollaborationSession).where( CollaborationSession.file_id == item.id, CollaborationSession.generation == item.generation, ).order_by(CollaborationSession.created_at.desc()) ) if session: should_recover = session.status == "ERROR" if session.status == "ACTIVE": created_at = session.created_at if created_at.tzinfo is None: created_at = created_at.replace(tzinfo=timezone.utc) if ( session.last_callback_at is None and datetime.now(timezone.utc) - created_at < timedelta(seconds=15) ): # The editor config can be requested twice before the first # browser has connected and emitted status 1. Keep a short # connection grace period so the second request does not retire # the freshly issued key as a false stale session. return session live_users = await _document_server_users(session.document_key) if live_users: return session # A service restart or rejected final callback can leave the row # ACTIVE after Document Server has already retired the editing # process. Reusing that key opens its forgotten copy as an # unbound server backup, so retire it exactly like a final callback. should_recover = True # A final callback ends the editing lifecycle for this key. Never # reactivate it: Document Server can retain a cached or forgotten copy # for the old key, especially across a container restart. recovered = ( await _recover_forgotten_content(session.document_key, item.file_type) if should_recover else None ) if recovered is not None: revision, _ = await collaboration_service.append_revision( db, item, recovered, source="SERVER_RECOVERY", created_by=user_id, change_summary="自动恢复在线文档服务器备份", ) session.base_revision_id = revision.id # The retired key can still point to Document Server's cache. A new # generation must use a new key or a later open can fall back to the # same server-side copy again. item.generation += 1 session.status = "RECOVERED" if recovered is not None else "CLOSED" session.closed_at = datetime.now(timezone.utc) await db.commit() session = None if not item.current_revision_id: raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="协作文件尚无可编辑内容") session = CollaborationSession( file_id=item.id, base_revision_id=item.current_revision_id, document_key=collaboration_document_key(item.id, item.generation), generation=item.generation, started_by=user_id, ) db.add(session) await db.commit() await db.refresh(session) return session async def build_editor_config( db: AsyncSession, item: CollaborationFile, user ) -> CollaborationEditorConfigRead: if ledger_access.is_ledger(item): await ledger_access.require_access(db, item, user) await onlyoffice_service.ensure_onlyoffice_available() revision = await db.get(CollaborationRevision, item.current_revision_id) if not revision or not Path(revision.file_uri).exists(): raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="协作文件内容不存在") can_edit = await collaboration_service.can_edit_file(db, item, user) can_request_edit = await collaboration_service.can_request_edit_file(db, item, user) can_download = await collaboration_service.can_export_file(db, item, user) can_save_as = can_download and await collaboration_service.can_create_file(db, item, user) session = await _active_session(db, item, user.id) now = datetime.now(timezone.utc) expires_at = now + timedelta(seconds=settings.ONLYOFFICE_CONFIG_TTL_SECONDS) config: dict[str, Any] = { "type": "desktop", "documentType": item.file_type, "document": { "fileType": item.extension, "key": session.document_key, "title": item.title, "url": _content_url(session.id), "permissions": { "chat": False, "copy": can_download, "comment": can_edit, "download": can_download, # In view mode ONLYOFFICE displays "Edit current file" only # when edit=true and onRequestEditRights is registered. CTMS # handles that event as an approval request, not an escalation. "edit": can_edit or can_request_edit, "fillForms": False, "modifyContentControl": can_edit, "modifyFilter": can_edit, "print": can_download, "protect": False, "review": False, }, }, "editorConfig": { "callbackUrl": _callback_url(session.id), "coEditing": {"mode": "fast", "change": False}, "customization": { "autosave": True, "chat": False, "comments": can_edit, "forcesave": can_edit, "help": False, "plugins": False, }, "lang": "zh-CN", "mode": "edit" if can_edit else "view", "user": {"id": str(user.id), "name": user.full_name}, }, } config["token"] = jwt.encode( {**config, "iat": int(now.timestamp()), "exp": int(expires_at.timestamp())}, settings.ONLYOFFICE_JWT_SECRET or "", algorithm="HS256", ) return CollaborationEditorConfigRead( file_id=item.id, file_name=item.title, access_mode="edit" if can_edit else "view", can_save_as=can_save_as, can_download=can_download, can_request_edit=can_request_edit, expires_at=expires_at, config=config, ) async def build_shared_editor_config( db: AsyncSession, item: CollaborationFile, link: CollaborationShareLink, *, client_id: str, display_name: str, ) -> CollaborationEditorConfigRead: if ledger_access.is_ledger(item): version = link.token_version item = await db.scalar(select(CollaborationFile).where( CollaborationFile.id == item.id, ).with_for_update().execution_options(populate_existing=True)) await db.refresh(link) if (not item or item.status != "ACTIVE" or not link.enabled or link.token_version != version or (link.expires_at and link.expires_at <= datetime.now(timezone.utc))): raise HTTPException(404, "共享链接不存在或已失效") await onlyoffice_service.ensure_onlyoffice_available() revision = await db.get(CollaborationRevision, item.current_revision_id) if not revision or not Path(revision.file_uri).exists(): raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="共享文件内容不存在") can_edit = link.access_mode == "EDIT" session = await _active_session(db, item, item.owner_id) now = datetime.now(timezone.utc) expires_at = now + timedelta(seconds=settings.ONLYOFFICE_CONFIG_TTL_SECONDS) if link.expires_at and link.expires_at < expires_at: expires_at = link.expires_at external_user_id = f"share-{link.id.hex[:12]}-{client_id[:32]}" if ledger_access.is_ledger(item): known_users = json.loads(session.active_users or "[]") session.active_users = json.dumps(list(dict.fromkeys([*known_users, external_user_id]))) await db.commit() config: dict[str, Any] = { "type": "desktop", "documentType": item.file_type, "document": { "fileType": item.extension, "key": session.document_key, "title": item.title, "url": _content_url(session.id), "permissions": { "chat": False, "copy": item.allow_export, "comment": can_edit, "download": item.allow_export, "edit": can_edit, "fillForms": False, "modifyContentControl": can_edit, "modifyFilter": can_edit, "print": item.allow_export, "protect": False, "review": False, }, }, "editorConfig": { "callbackUrl": _callback_url(session.id), "coEditing": {"mode": "fast", "change": False}, "customization": { "autosave": can_edit, "chat": False, "comments": can_edit, "forcesave": can_edit, "help": False, "plugins": False, }, "lang": "zh-CN", "mode": "edit" if can_edit else "view", "user": {"id": external_user_id, "name": display_name}, }, } config["token"] = jwt.encode( {**config, "iat": int(now.timestamp()), "exp": int(expires_at.timestamp())}, settings.ONLYOFFICE_JWT_SECRET or "", algorithm="HS256", ) return CollaborationEditorConfigRead( file_id=item.id, file_name=item.title, access_mode="edit" if can_edit else "view", can_save_as=False, can_download=item.allow_export, expires_at=expires_at, config=config, ) async def get_session_content( db: AsyncSession, session_id: uuid.UUID, authorization: str | None ) -> tuple[CollaborationRevision, CollaborationFile]: session = await db.get(CollaborationSession, session_id) if not session: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="协作会话不存在") onlyoffice_service.validate_outbox_token(authorization, _content_url(session_id)) revision = await db.get(CollaborationRevision, session.base_revision_id) item = await db.get(CollaborationFile, session.file_id) if not revision or not item or not Path(revision.file_uri).exists(): raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="协作文件内容不存在") if ledger_access.is_ledger(item) and session.generation != item.generation: raise HTTPException(403, "台账会话已失效,请重新打开") return revision, item def validate_callback_token(token: str | None, payload: CollaborationCallbackPayload) -> dict[str, Any]: if not token: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="缺少 ONLYOFFICE 回调签名") value = token.strip() if " " in value: scheme, credential = value.split(" ", 1) if scheme.lower() != "bearer": raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="ONLYOFFICE 回调签名格式无效") value = credential.strip() try: decoded = onlyoffice_service.decode_onlyoffice_token(value) except JWTError as exc: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="ONLYOFFICE 回调签名无效") from exc signed = decoded.get("payload") if isinstance(decoded.get("payload"), dict) else decoded if not isinstance(signed, dict): raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="ONLYOFFICE 回调载荷无效") signed_key = signed.get("key") signed_status = signed.get("status") if not isinstance(signed_key, str) or not hmac.compare_digest(signed_key, payload.key): raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="ONLYOFFICE 回调 key 不匹配") if not isinstance(signed_status, int) or signed_status != payload.status: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="ONLYOFFICE 回调状态不匹配") if payload.url: signed_url = signed.get("url") if not isinstance(signed_url, str) or not hmac.compare_digest(signed_url, payload.url): raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="ONLYOFFICE 回调文件地址不匹配") return decoded def _callback_fingerprint(payload: CollaborationCallbackPayload) -> str: normalized = payload.model_dump(mode="json", exclude_none=True) return hashlib.sha256(json.dumps(normalized, sort_keys=True, separators=(",", ":")).encode()).hexdigest() def _url_origin_matches(actual, expected) -> bool: actual_port = actual.port or (443 if actual.scheme == "https" else 80) expected_port = expected.port or (443 if expected.scheme == "https" else 80) return ( actual.scheme == expected.scheme and actual.hostname and actual.hostname.lower() == (expected.hostname or "").lower() and actual_port == expected_port ) def _validate_result_url(url: str) -> str: actual = urlsplit(url) if ( actual.scheme not in {"http", "https"} or actual.username or actual.password or actual.fragment or not actual.hostname ): raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="ONLYOFFICE 保存地址不受信任") internal = urlsplit(settings.ONLYOFFICE_INTERNAL_URL.rstrip("/")) if _url_origin_matches(actual, internal): return urlunsplit((internal.scheme, internal.netloc, actual.path, actual.query, "")) public = urlsplit(settings.FRONTEND_PUBLIC_URL.rstrip("/")) proxy_prefix = "/onlyoffice/" if not _url_origin_matches(actual, public) or not actual.path.startswith(proxy_prefix): raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="ONLYOFFICE 保存地址不受信任") internal_path = f"{internal.path.rstrip('/')}/{actual.path[len(proxy_prefix):]}" return urlunsplit((internal.scheme, internal.netloc, internal_path, actual.query, "")) async def _download_result(url: str) -> bytes: download_url = _validate_result_url(url) try: async with httpx.AsyncClient(timeout=30.0, follow_redirects=False) as client: async with client.stream("GET", download_url) as response: if response.status_code != status.HTTP_200_OK: raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="ONLYOFFICE 保存文件下载失败") content = bytearray() async for chunk in response.aiter_bytes(): content.extend(chunk) if len(content) > settings.COLLABORATION_MAX_FILE_BYTES: raise HTTPException(status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, detail="ONLYOFFICE 保存文件超出限制") except httpx.HTTPError as exc: raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="ONLYOFFICE 保存文件下载失败") from exc return bytes(content) def _validate_recovered_content(content: bytes, file_type: str) -> None: required_part = { "word": "word/document.xml", "cell": "xl/workbook.xml", "slide": "ppt/presentation.xml", }.get(file_type) if not required_part or not zipfile.is_zipfile(io.BytesIO(content)): raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="在线文档服务器备份格式无效") try: with zipfile.ZipFile(io.BytesIO(content)) as package: if required_part not in package.namelist() or package.testzip() is not None: raise ValueError except (zipfile.BadZipFile, ValueError) as exc: raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="在线文档服务器备份已损坏") from exc async def _recover_forgotten_content(document_key: str, file_type: str) -> bytes | None: """Download a Document Server backup left behind by a failed final save.""" command = {"c": "getForgotten", "key": document_key} token = jwt.encode(command, settings.ONLYOFFICE_JWT_SECRET or "", algorithm="HS256") command_url = f"{settings.ONLYOFFICE_INTERNAL_URL.rstrip('/')}/command" try: async with httpx.AsyncClient(timeout=10.0, follow_redirects=False) as client: response = await client.post( command_url, params={"shardkey": document_key}, json={**command, "token": token}, ) if response.status_code != status.HTTP_200_OK: raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="在线文档备份恢复服务暂不可用") payload = response.json() except (httpx.HTTPError, ValueError) as exc: raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="在线文档备份恢复服务暂不可用") from exc error = payload.get("error") if isinstance(payload, dict) else None if error == 1: # Document Server no longer has a forgotten copy. Starting from the # last confirmed CTMS revision is then the only recoverable state. return None if ( error != 0 or payload.get("key") != document_key or not isinstance(payload.get("url"), str) or not payload["url"] ): raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="在线文档服务器返回的备份信息无效") content = await _download_result(payload["url"]) _validate_recovered_content(content, file_type) return content async def _document_server_users(document_key: str) -> list[str]: """Return live editor ids for a key without trusting stale database state.""" command = {"c": "info", "key": document_key} token = jwt.encode(command, settings.ONLYOFFICE_JWT_SECRET or "", algorithm="HS256") command_url = f"{settings.ONLYOFFICE_INTERNAL_URL.rstrip('/')}/command" try: async with httpx.AsyncClient(timeout=10.0, follow_redirects=False) as client: response = await client.post( command_url, params={"shardkey": document_key}, json={**command, "token": token}, ) if response.status_code != status.HTTP_200_OK: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="在线文档会话检查服务暂不可用", ) payload = response.json() except (httpx.HTTPError, ValueError) as exc: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="在线文档会话检查服务暂不可用", ) from exc error = payload.get("error") if isinstance(payload, dict) else None if error == 1: # The database can retain an ACTIVE row after an interrupted callback, # while Document Server no longer has a live editing process for it. return [] users = payload.get("users") if isinstance(payload, dict) else None if ( error != 0 or payload.get("key") != document_key or not isinstance(users, list) or any(not isinstance(user_id, str) or not user_id for user_id in users) ): raise HTTPException( status_code=status.HTTP_502_BAD_GATEWAY, detail="在线文档服务器返回的会话信息无效", ) return list(dict.fromkeys(users)) async def list_live_editing_sessions(db: AsyncSession) -> list[tuple[str, int]]: """List file titles and live editor counts for deployment safety checks.""" rows = ( await db.execute( select(CollaborationSession.document_key, CollaborationFile.title) .join(CollaborationFile, CollaborationFile.id == CollaborationSession.file_id) .where(CollaborationSession.status == "ACTIVE") .order_by(CollaborationFile.title) ) ).all() active: list[tuple[str, int]] = [] for document_key, title in rows: users = await _document_server_users(document_key) if users: active.append((title, len(users))) return active async def _callback_user( db: AsyncSession, payload: CollaborationCallbackPayload, session: CollaborationSession ) -> User | None: has_public_share_user = False for value in payload.users or json.loads(session.active_users or "[]"): if value.startswith("share-"): has_public_share_user = True continue try: user = await db.get(User, uuid.UUID(value)) except (ValueError, TypeError): user = None if user: return user if has_public_share_user: return None user = await db.get(User, session.started_by) if not user: raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="协作会话用户不存在") return user async def _ledger_callback_can_edit(db, item, payload, session): users = payload.users or json.loads(session.active_users or "[]") if not users: return await ledger_access.can_edit(db, item, await _callback_user(db, payload, session)) for value in users: try: actor = await db.get(User, uuid.UUID(value)) except (ValueError, TypeError): continue if await ledger_access.can_edit(db, item, actor): return True link = await db.scalar(select(CollaborationShareLink).where( CollaborationShareLink.file_id == item.id, CollaborationShareLink.enabled.is_(True), CollaborationShareLink.access_mode == "EDIT", )) if not link or (link.expires_at and link.expires_at <= datetime.now(timezone.utc)): return False prefix = f"share-{link.id.hex[:12]}-" return any(value.startswith(prefix) for value in users) async def process_callback( db: AsyncSession, session_id: uuid.UUID, payload: CollaborationCallbackPayload, ) -> dict[str, int]: # Lock the file before writing sessions, matching editor initialization and # ledger permission changes. This also serializes callbacks across generations. session = await db.scalar( select(CollaborationSession) .join(CollaborationFile, CollaborationFile.id == CollaborationSession.file_id) .where(CollaborationSession.id == session_id) .with_for_update(of=CollaborationFile) ) if not session: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="协作会话不存在") if not hmac.compare_digest(session.document_key, payload.key): raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="协作会话 key 不匹配") fingerprint = _callback_fingerprint(payload) duplicate = await db.scalar(select(CollaborationCallbackReceipt.id).where( CollaborationCallbackReceipt.session_id == session.id, CollaborationCallbackReceipt.fingerprint == fingerprint, )) if duplicate: return {"error": 0} item = await db.get(CollaborationFile, session.file_id) if not item: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="协作文件不存在") if ledger_access.is_ledger(item): # Same lock as grant/settings changes: a stale callback cannot race a revocation. item = await db.scalar(select(CollaborationFile).where( CollaborationFile.id == item.id, ).with_for_update().execution_options(populate_existing=True)) session.last_callback_at = datetime.now(timezone.utc) result = "ACKNOWLEDGED" saved_revision_id = None if payload.status in {2, 6}: if not payload.url: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="ONLYOFFICE 保存回调缺少文件地址") if session.generation != item.generation: result = "STALE" elif ledger_access.is_ledger(item) and ( item.status != "ACTIVE" or not await _ledger_callback_can_edit(db, item, payload, session) ): result = "ACCESS_REVOKED" else: content = await _download_result(payload.url) actor = await _callback_user(db, payload, session) source = "SESSION_CLOSE" if payload.status == 2 else "FORCE_SAVE" if actor is None: source = "SHARE_SESSION_CLOSE" if payload.status == 2 else "SHARE_FORCE_SAVE" revision, created = await collaboration_service.append_revision( db, item, content, source=source, created_by=actor.id if actor else None ) saved_revision_id = revision.id result = "SAVED" if created else "UNCHANGED" if payload.status == 2: item.generation += 1 session.status = "CLOSED" session.closed_at = datetime.now(timezone.utc) else: # 强制保存不结束当前共同编辑会话;同步基线可保证 Document # Server 缓存重建时仍从最近一次持久化内容恢复。 session.base_revision_id = revision.id elif payload.status == 4: session.status = "CLOSED" session.closed_at = datetime.now(timezone.utc) result = "UNCHANGED" elif payload.status == 3: session.status = "ERROR" result = "ERROR" elif payload.status == 7: # A force-save error does not close the live co-editing session. The # final status 2 callback can still persist the document normally. result = "ERROR" if payload.users or not ledger_access.is_ledger(item): session.active_users = json.dumps(payload.users, ensure_ascii=True) db.add(CollaborationCallbackReceipt( session_id=session.id, fingerprint=fingerprint, callback_status=payload.status, result=result, revision_id=saved_revision_id, )) await db.commit() # ONLYOFFICE 要求回调处理器在接收并记录状态后固定确认成功。 # status 3/7 表示文档服务自身保存失败,不应通过 error=1 制造重试环。 return {"error": 0}