307 lines
14 KiB
Python
307 lines
14 KiB
Python
"""App-layer publish-time resource checks for workflow graphs.
|
||
|
||
The harness validator stays store-agnostic via its ``resource_checker`` hook;
|
||
this module is where the Gateway binds that hook to the real agent / skill /
|
||
data-source / published-version stores. It pre-fetches the catalogs the caller
|
||
is allowed to see, then exposes a **sync** checker so it can be passed straight
|
||
into ``validate_workflow_graph``.
|
||
|
||
Never imports from ``packages/harness`` internals beyond the public workflow
|
||
API, and the harness never imports this module.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass, field
|
||
from typing import Any
|
||
|
||
from fastapi import Request
|
||
|
||
from app.gateway.routers._workflow_planner_seed import WORKFLOW_PLANNER_AGENT_ID
|
||
from deerflow.config.workflow_config import WorkflowConfig
|
||
from deerflow.workflows.errors import WorkflowError
|
||
from deerflow.workflows.security.sql_policy import assert_read_only, assert_tables_allowed
|
||
|
||
# Header-ish keys that must never be configured as literal values on the canvas:
|
||
# credentials belong in an http data source referenced by ``credentialRef``.
|
||
_LITERAL_SECRET_KEYS = frozenset(
|
||
{
|
||
"authorization",
|
||
"proxy-authorization",
|
||
"cookie",
|
||
"x-api-key",
|
||
"apikey",
|
||
"api_key",
|
||
"secret",
|
||
"client_secret",
|
||
"password",
|
||
"access_token",
|
||
"refresh_token",
|
||
}
|
||
)
|
||
|
||
_HTTP_METHODS = frozenset({"GET", "POST", "PUT", "PATCH", "DELETE"})
|
||
|
||
|
||
@dataclass
|
||
class ResourceCatalog:
|
||
"""Everything the checker is allowed to reference, prefetched."""
|
||
|
||
agent_ids: set[str] = field(default_factory=set)
|
||
# agent_id -> cached skill whitelist from the agent store (empty = no
|
||
# restriction recorded).
|
||
agent_skills: dict[str, set[str]] = field(default_factory=dict)
|
||
enabled_skills: set[str] = field(default_factory=set)
|
||
# data source id -> public row (kind / enabled / allowed_* / max_rows).
|
||
data_sources: dict[str, dict[str, Any]] = field(default_factory=dict)
|
||
# workflow_id -> set of published version ids.
|
||
published_subworkflows: dict[str, set[str]] = field(default_factory=dict)
|
||
|
||
|
||
def _issue(message: str, field_path: str) -> dict[str, Any]:
|
||
return {"message": message, "field": field_path}
|
||
|
||
|
||
async def build_resource_catalog(request: Request, *, user_id: str) -> ResourceCatalog:
|
||
catalog = ResourceCatalog()
|
||
|
||
agent_store = getattr(request.app.state, "agent_store", None)
|
||
if agent_store is not None:
|
||
try:
|
||
if hasattr(agent_store, "list_visible"):
|
||
rows = await agent_store.list_visible(user_id)
|
||
elif hasattr(agent_store, "list_all"):
|
||
rows = await agent_store.list_all()
|
||
else:
|
||
rows = []
|
||
catalog.agent_ids = {
|
||
str(row.get("id") or row.get("agent_id") or "")
|
||
for row in rows or []
|
||
if isinstance(row, dict)
|
||
and (row.get("id") or row.get("agent_id"))
|
||
and str(row.get("id") or row.get("agent_id") or "") != WORKFLOW_PLANNER_AGENT_ID
|
||
}
|
||
if hasattr(agent_store, "get_extras_for"):
|
||
extras = await agent_store.get_extras_for([aid for aid in catalog.agent_ids if aid])
|
||
for agent_id, extra in (extras or {}).items():
|
||
skills = (extra or {}).get("skills") or []
|
||
if skills:
|
||
catalog.agent_skills[str(agent_id)] = {str(s) for s in skills}
|
||
except Exception: # noqa: BLE001 - publish checks degrade to strict-by-empty
|
||
catalog.agent_ids = set()
|
||
|
||
try:
|
||
from deerflow.skills import get_or_new_skill_storage
|
||
|
||
catalog.enabled_skills = {
|
||
str(getattr(skill, "name"))
|
||
for skill in get_or_new_skill_storage().load_skills(enabled_only=True) or []
|
||
if getattr(skill, "name", None)
|
||
}
|
||
except Exception: # noqa: BLE001
|
||
catalog.enabled_skills = set()
|
||
|
||
ds_store = getattr(request.app.state, "workflow_data_source_store", None)
|
||
if ds_store is not None:
|
||
try:
|
||
for row in await ds_store.list_sources(owner_id=user_id, include_shared=True) or []:
|
||
catalog.data_sources[str(row.get("id"))] = row
|
||
except Exception: # noqa: BLE001
|
||
catalog.data_sources = {}
|
||
|
||
wf_store = getattr(request.app.state, "workflow_store", None)
|
||
if wf_store is not None:
|
||
try:
|
||
for row in await wf_store.list_published_summaries() or []:
|
||
versions = catalog.published_subworkflows.setdefault(str(row.get("workflow_id")), set())
|
||
versions.add(str(row.get("version_id")))
|
||
except Exception: # noqa: BLE001
|
||
catalog.published_subworkflows = {}
|
||
|
||
return catalog
|
||
|
||
|
||
def _check_agent(config: dict[str, Any], catalog: ResourceCatalog) -> dict[str, Any] | None:
|
||
# Empty agentId means the runtime's built-in default agent — a real,
|
||
# executable fallback, not missing configuration.
|
||
agent_id = str(config.get("agentId") or config.get("agent_id") or "") or "default"
|
||
if agent_id != "default" and agent_id not in catalog.agent_ids:
|
||
return _issue("未选择可访问的智能体", "config.agentId")
|
||
if not str(config.get("promptTemplate") or config.get("prompt_template") or "").strip():
|
||
return _issue("智能体节点的提示词不能为空", "config.promptTemplate")
|
||
mode = str(config.get("responseMode") or config.get("response_mode") or "text")
|
||
schema = config.get("responseSchema") or config.get("response_schema")
|
||
if mode == "json":
|
||
if not isinstance(schema, dict) or not schema:
|
||
return _issue("JSON 输出模式必须配置 responseSchema 对象", "config.responseSchema")
|
||
if schema.get("type") not in (None, "object"):
|
||
return _issue("responseSchema 目前仅支持 object 类型", "config.responseSchema")
|
||
return None
|
||
|
||
|
||
def _check_skill(config: dict[str, Any], catalog: ResourceCatalog) -> dict[str, Any] | None:
|
||
mode = str(config.get("mode") or "agent_skill")
|
||
if mode != "agent_skill":
|
||
return _issue("callable_skill 模式暂未开放执行，请改用 agent_skill", "config.mode")
|
||
agent_id = str(config.get("agentId") or config.get("agent_id") or "") or "default"
|
||
if agent_id != "default" and agent_id not in catalog.agent_ids:
|
||
return _issue("技能节点引用的智能体不可访问", "config.agentId")
|
||
skill_names = [str(s) for s in (config.get("skillNames") or config.get("skill_names") or [])]
|
||
if not skill_names:
|
||
return _issue("技能节点至少需要选择一个技能", "config.skillNames")
|
||
for name in skill_names:
|
||
if name not in catalog.enabled_skills:
|
||
return _issue(f"技能不存在或未启用：{name}", "config.skillNames")
|
||
whitelist = catalog.agent_skills.get(agent_id)
|
||
if whitelist:
|
||
missing = [name for name in skill_names if name not in whitelist]
|
||
if missing:
|
||
return _issue(f"技能不在智能体白名单内：{'、'.join(missing)}", "config.skillNames")
|
||
return None
|
||
|
||
|
||
def _check_sql(config: dict[str, Any], catalog: ResourceCatalog, cfg: WorkflowConfig) -> dict[str, Any] | None:
|
||
source_id = str(config.get("dataSourceId") or config.get("data_source_id") or "")
|
||
if not source_id:
|
||
return _issue("SQL 节点未选择数据源", "config.dataSourceId")
|
||
source = catalog.data_sources.get(source_id)
|
||
if source is None:
|
||
return _issue("数据源不存在或无权访问", "config.dataSourceId")
|
||
if not source.get("enabled", True):
|
||
return _issue("数据源已被禁用", "config.dataSourceId")
|
||
if str(source.get("kind") or "sql") != "sql":
|
||
return _issue("该资源不是 SQL 数据源", "config.dataSourceId")
|
||
statement = str(config.get("statement") or config.get("queryTemplate") or config.get("query_template") or "")
|
||
if not statement.strip():
|
||
return _issue("SQL 节点缺少查询语句", "config.statement")
|
||
try:
|
||
assert_read_only(statement)
|
||
assert_tables_allowed(statement, list(source.get("allowed_tables") or []))
|
||
except WorkflowError as exc:
|
||
return _issue(str(exc.message if hasattr(exc, "message") else exc), "config.statement")
|
||
parameters = config.get("parameters")
|
||
if parameters is not None and not isinstance(parameters, dict):
|
||
return _issue("SQL 参数必须是对象", "config.parameters")
|
||
max_rows = config.get("maxRows") or config.get("max_rows")
|
||
if max_rows is not None:
|
||
try:
|
||
rows = int(max_rows)
|
||
except (TypeError, ValueError):
|
||
return _issue("maxRows 必须是整数", "config.maxRows")
|
||
ceiling = min(int(cfg.sql.max_rows), int(source.get("max_rows") or cfg.sql.max_rows))
|
||
if rows < 1 or rows > ceiling:
|
||
return _issue(f"maxRows 超过上限 {ceiling}", "config.maxRows")
|
||
return None
|
||
|
||
|
||
def _check_http(config: dict[str, Any], catalog: ResourceCatalog) -> dict[str, Any] | None:
|
||
method = str(config.get("method") or "GET").upper()
|
||
if method not in _HTTP_METHODS:
|
||
return _issue(f"不支持的 HTTP 方法：{method}", "config.method")
|
||
headers = config.get("headers")
|
||
if headers is not None:
|
||
if not isinstance(headers, dict):
|
||
return _issue("请求头必须是字符串到字符串的映射", "config.headers")
|
||
for name in headers:
|
||
if str(name).strip().lower() in _LITERAL_SECRET_KEYS:
|
||
return _issue(f"请勿在节点上明文配置 {name}，改用凭据引用", "config.headers")
|
||
url = str(config.get("url") or "").strip()
|
||
credential_ref = str(config.get("credentialRef") or config.get("credential_ref") or "").strip()
|
||
if not url and not credential_ref:
|
||
return _issue("HTTP 节点缺少 URL 或凭据引用", "config.url")
|
||
if credential_ref:
|
||
source = catalog.data_sources.get(credential_ref)
|
||
if source is None:
|
||
return _issue("引用的凭据（HTTP 数据源）不存在或无权访问", "config.credentialRef")
|
||
if not source.get("enabled", True):
|
||
return _issue("引用的凭据已被禁用", "config.credentialRef")
|
||
if str(source.get("kind") or "") != "http":
|
||
return _issue("引用的凭据不是 HTTP 类型数据源", "config.credentialRef")
|
||
allowed = {str(m).upper() for m in (source.get("allowed_methods") or [])}
|
||
if allowed and method not in allowed:
|
||
return _issue(f"方法 {method} 不在凭据资源允许的方法列表内", "config.method")
|
||
return None
|
||
|
||
|
||
def _check_code(config: dict[str, Any], cfg: WorkflowConfig) -> dict[str, Any] | None:
|
||
if not cfg.code.enabled:
|
||
return _issue("代码节点未启用（需管理员在 config.yaml 中开启 workflows.code.enabled）", "")
|
||
language = config.get("language")
|
||
if language is not None and str(language) != "python":
|
||
return _issue("代码节点目前仅支持 python", "config.language")
|
||
source = str(config.get("source") or "")
|
||
if not source.strip():
|
||
return _issue("代码节点没有源代码", "config.source")
|
||
timeout = config.get("timeoutSeconds") or config.get("timeout_seconds")
|
||
if timeout is not None:
|
||
try:
|
||
seconds = int(timeout)
|
||
except (TypeError, ValueError):
|
||
return _issue("timeoutSeconds 必须是整数", "config.timeoutSeconds")
|
||
if seconds < 1 or seconds > int(cfg.code.timeout_seconds):
|
||
return _issue(f"timeoutSeconds 超过上限 {cfg.code.timeout_seconds}", "config.timeoutSeconds")
|
||
return None
|
||
|
||
|
||
def _check_output(config: dict[str, Any]) -> dict[str, Any] | None:
|
||
mapping = config.get("mapping")
|
||
if mapping is None:
|
||
return None
|
||
if not isinstance(mapping, dict):
|
||
return _issue("输出映射必须是对象", "config.mapping")
|
||
for key, template in mapping.items():
|
||
if not isinstance(template, str) or not template.strip():
|
||
return _issue(f"输出字段 {key} 的映射模板为空", f"config.mapping.{key}")
|
||
return None
|
||
|
||
|
||
def _check_subworkflow(config: dict[str, Any], catalog: ResourceCatalog, *, current_workflow_id: str) -> dict[str, Any] | None:
|
||
workflow_id = str(config.get("workflowId") or config.get("workflow_id") or "")
|
||
if not workflow_id:
|
||
return _issue("子工作流节点缺少 workflowId", "config.workflowId")
|
||
version_id = str(config.get("versionId") or config.get("version_id") or "")
|
||
if not version_id:
|
||
return _issue("子工作流节点缺少 versionId", "config.versionId")
|
||
if workflow_id == current_workflow_id:
|
||
return _issue("子工作流不能引用自身", "config.workflowId")
|
||
versions = catalog.published_subworkflows.get(workflow_id)
|
||
if not versions:
|
||
return _issue("子工作流未发布或无权访问", "config.workflowId")
|
||
if version_id not in versions:
|
||
return _issue("子工作流版本不存在或已下线", "config.versionId")
|
||
return None
|
||
|
||
|
||
def make_resource_checker(
|
||
catalog: ResourceCatalog,
|
||
*,
|
||
config: WorkflowConfig,
|
||
current_workflow_id: str = "",
|
||
):
|
||
"""Build the sync ``resource_checker`` consumed by ``validate_workflow_graph``."""
|
||
|
||
def checker(node_type: str, node_config: dict[str, Any]):
|
||
try:
|
||
if node_type == "agent":
|
||
return _check_agent(node_config, catalog)
|
||
if node_type == "skill":
|
||
return _check_skill(node_config, catalog)
|
||
if node_type == "sql_read":
|
||
return _check_sql(node_config, catalog, config)
|
||
if node_type == "http":
|
||
return _check_http(node_config, catalog)
|
||
if node_type == "code":
|
||
return _check_code(node_config, config)
|
||
if node_type == "output":
|
||
return _check_output(node_config)
|
||
if node_type == "subworkflow":
|
||
return _check_subworkflow(node_config, catalog, current_workflow_id=current_workflow_id)
|
||
except Exception: # noqa: BLE001 - a broken check must fail closed
|
||
return _issue("资源配置校验失败，请检查节点配置", "")
|
||
return None
|
||
|
||
return checker
|
||
|
||
|
||
__all__ = ["ResourceCatalog", "build_resource_catalog", "make_resource_checker"]
|