zkxw/backend/app/services/workflow_callback.py

"""工作流回调服务 —— 流程状态变更时通知业务系统"""
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,
    )