Cheng Zhou
c1dd8def2c
新增与项目平级的协作台账，提供空白模板、独立账号授权和在线编辑。 复用共享库的访问权限与历史版本界面，拆分信息维护入口并统一工作台卡片样式。 补充在线文档字体构建支持、数据库表结构升级和回归测试。 不包含本地台账内容、账号授权记录或历史文件迁移数据。
472 lines
20 KiB
Python
472 lines
20 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import uuid
|
|
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:
|
|
if session.status != "ACTIVE":
|
|
session.status = "ACTIVE"
|
|
session.closed_at = None
|
|
await db.commit()
|
|
await db.refresh(session)
|
|
return session
|
|
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)
|
|
|
|
|
|
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 in {3, 7}:
|
|
session.status = "ERROR"
|
|
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}
|