106 lines
3.8 KiB
Python
106 lines
3.8 KiB
Python
"""工作流回调服务 —— 流程状态变更时通知业务系统"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import TYPE_CHECKING, Any, Callable, Dict, Optional
|
|
|
|
if TYPE_CHECKING:
|
|
from app.models.models import WfProcessInstance
|
|
|
|
logger = logging.getLogger("sqxp.workflow_callback")
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 本地回调注册表：内部路径（以 / 开头）直接在进程内调用，不走 HTTP
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_LocalCallbackFn = Callable[[Any, str, str], None] # (db, business_id, event) -> None
|
|
|
|
_LOCAL_CALLBACKS: Dict[str, _LocalCallbackFn] = {}
|
|
|
|
|
|
def register_local_callback(path: str, fn: _LocalCallbackFn) -> None:
|
|
"""注册本地回调函数。当 callback_url 以 path 开头时直接调用 fn。"""
|
|
_LOCAL_CALLBACKS[path] = fn
|
|
|
|
|
|
def _dispatch_local(db: Any, instance: WfProcessInstance, event: str) -> bool:
|
|
"""尝试本地直调。返回 True 表示已处理，False 表示无匹配的本地回调。"""
|
|
url = instance.callback_url or ""
|
|
if not url.startswith("/"):
|
|
return False
|
|
for prefix, fn in _LOCAL_CALLBACKS.items():
|
|
if url.startswith(prefix):
|
|
try:
|
|
fn(db, instance.business_id or "", event)
|
|
logger.info(
|
|
"工作流本地回调成功: event=%s instance=%s path=%s",
|
|
event,
|
|
instance.id,
|
|
url,
|
|
)
|
|
return True
|
|
except Exception:
|
|
logger.exception(
|
|
"工作流本地回调失败: event=%s instance=%s path=%s",
|
|
event,
|
|
instance.id,
|
|
url,
|
|
)
|
|
return True # 已匹配但执行失败，不再走 HTTP
|
|
return False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# trigger_callback（对外入口）
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def validate_callback_url(callback_url: Optional[str]) -> Optional[str]:
|
|
"""校验并规范化 callback_url。
|
|
|
|
仅允许空值或已注册的本地内部路径（以 / 开头）。
|
|
禁止 http(s) 等外链，避免 SSRF。
|
|
"""
|
|
url = (callback_url or "").strip()
|
|
if not url:
|
|
return None
|
|
lower = url.lower()
|
|
if lower.startswith("http://") or lower.startswith("https://") or "://" in url:
|
|
raise ValueError("callbackUrl 仅支持已注册的本地内部路径，禁止 HTTP(S) 外链")
|
|
if not url.startswith("/"):
|
|
raise ValueError("callbackUrl 必须为以 / 开头的本地内部路径")
|
|
matched = any(url.startswith(prefix) for prefix in _LOCAL_CALLBACKS)
|
|
if not matched:
|
|
raise ValueError("callbackUrl 未注册为本地回调，禁止使用")
|
|
return url
|
|
|
|
|
|
def trigger_callback(db: Any, instance: WfProcessInstance, event: str) -> None:
|
|
"""流程实例状态变更后，向业务系统发送回调通知。
|
|
|
|
仅支持已注册的本地内部路径（进程内直调）。不再发起任意 HTTP 回调，避免 SSRF。
|
|
"""
|
|
if not instance.callback_url:
|
|
return
|
|
|
|
url = (instance.callback_url or "").strip()
|
|
lower = url.lower()
|
|
if lower.startswith("http://") or lower.startswith("https://") or "://" in url:
|
|
logger.warning(
|
|
"拒绝工作流 HTTP 回调(SSRF 防护): event=%s instance=%s url=%s",
|
|
event,
|
|
instance.id,
|
|
url,
|
|
)
|
|
return
|
|
|
|
if _dispatch_local(db, instance, event):
|
|
return
|
|
|
|
logger.warning(
|
|
"工作流回调未匹配本地注册表，已跳过: event=%s instance=%s url=%s",
|
|
event,
|
|
instance.id,
|
|
url,
|
|
)
|