mirror of
https://github.com/langgenius/dify.git
synced 2026-11-11 22:02:23 +08:00
Charles Yao
231e423431
Track backend integration, Ruff cleanup and migration reconciliation in ENG-1271. Retain required copy contracts without implementing frontend interactions.
489 lines
18 KiB
Python
489 lines
18 KiB
Python
"""Pure admission for the installed, qualified JSON Process parse workflow.
|
|
|
|
This module inspects raw source before Graphon builds a Tool node. Its resolved
|
|
parameter envelope is an expectation for the later native adapter, never a
|
|
replacement for native parameter conversion or JSONPath evaluation.
|
|
"""
|
|
|
|
import re
|
|
from collections.abc import Mapping
|
|
from dataclasses import dataclass
|
|
from typing import Any, NoReturn
|
|
|
|
from core.workflow.variable_prefixes import (
|
|
CONVERSATION_VARIABLE_NODE_ID,
|
|
ENVIRONMENT_VARIABLE_NODE_ID,
|
|
SYSTEM_VARIABLE_NODE_ID,
|
|
)
|
|
from runtime_contracts.builder.execution import (
|
|
AdmittedNodeBinding,
|
|
BuilderExecutionPolicyError,
|
|
QualifiedJsonToolBinding,
|
|
canonical_digest,
|
|
)
|
|
from services.dify_builder.application.execution_policy import (
|
|
RestrictedAdmissionSnapshot,
|
|
admit_raw_execution_metadata,
|
|
)
|
|
from services.dify_builder.revision import node_config_digest
|
|
|
|
__all__ = [
|
|
"JsonToolAdmission",
|
|
"QualifiedJsonToolBinding",
|
|
"admit_json_workflow",
|
|
"resolve_json_parameters",
|
|
"validate_json_parameters",
|
|
]
|
|
|
|
_PLUGIN_ID = "langgenius/json_process"
|
|
_PROVIDER_ID = "langgenius/json_process/json_process"
|
|
_UNIQUE_IDENTIFIER = "langgenius/json_process:0.0.8@3333687ec0412e1f2a1798d632fc5f3187f36ebd4bf2432a599a96c74e92f503"
|
|
_NODE_ID = re.compile(r"[A-Za-z0-9_-]{1,128}\Z", re.ASCII)
|
|
_RESERVED_NATIVE_OWNERS = frozenset(
|
|
{SYSTEM_VARIABLE_NODE_ID, ENVIRONMENT_VARIABLE_NODE_ID, CONVERSATION_VARIABLE_NODE_ID}
|
|
)
|
|
_START_NAME = re.compile(r"[A-Za-z_][A-Za-z0-9_]{0,63}\Z", re.ASCII)
|
|
_PURE_MIXED = re.compile(r"\{\{#([A-Za-z0-9_]{1,50})\.([A-Za-z_][A-Za-z0-9_]{0,29})#\}\}\Z", re.ASCII)
|
|
_PATH_PART = re.compile(r"\.([A-Za-z_][A-Za-z0-9_]{0,63})|\[([0-9]{1,5}|\*)\]", re.ASCII)
|
|
_UNSAFE_CONTENT = re.compile(r"\.|\{\{#|#\}\}|\\u(?:002e|0023|007b|007d)", re.IGNORECASE)
|
|
_START_DATA = {"type", "version", "variables", "title", "desc", "selected"}
|
|
_END_DATA = {"type", "version", "outputs", "title", "desc", "selected"}
|
|
_TOOL_DATA = {
|
|
"type",
|
|
"version",
|
|
"tool_node_version",
|
|
"provider_type",
|
|
"provider_id",
|
|
"provider_name",
|
|
"plugin_id",
|
|
"plugin_unique_identifier",
|
|
"tool_name",
|
|
"tool_label",
|
|
"tool_description",
|
|
"tool_configurations",
|
|
"tool_parameters",
|
|
"paramSchemas",
|
|
"params",
|
|
"output_schema",
|
|
"title",
|
|
"desc",
|
|
"selected",
|
|
"credential_id",
|
|
"credential_type",
|
|
"credential_name",
|
|
"is_team_authorization",
|
|
"provider_show_name",
|
|
"provider_icon",
|
|
"provider_icon_dark",
|
|
}
|
|
_PARAM_SCHEMA_KEYS = {
|
|
"auto_generate",
|
|
"default",
|
|
"form",
|
|
"human_description",
|
|
"input_schema",
|
|
"label",
|
|
"llm_description",
|
|
"max",
|
|
"min",
|
|
"multiple",
|
|
"name",
|
|
"options",
|
|
"placeholder",
|
|
"precision",
|
|
"required",
|
|
"scope",
|
|
"template",
|
|
"type",
|
|
}
|
|
_GRAPH_KEYS = {"nodes", "edges", "viewport"}
|
|
_NODE_KEYS = {
|
|
"id",
|
|
"data",
|
|
"height",
|
|
"position",
|
|
"positionAbsolute",
|
|
"selected",
|
|
"sourcePosition",
|
|
"targetPosition",
|
|
"type",
|
|
"width",
|
|
"zIndex",
|
|
}
|
|
_EDGE_KEYS = {
|
|
"id",
|
|
"source",
|
|
"target",
|
|
"sourceHandle",
|
|
"targetHandle",
|
|
"type",
|
|
"data",
|
|
"zIndex",
|
|
}
|
|
_MAX_CONTENT = 65536
|
|
_MAX_FILTER = 1024
|
|
_MAX_PROVIDER_SHOW_NAME = 256
|
|
_MAX_PROVIDER_ICON = 2048
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class JsonToolAdmission:
|
|
nodes: tuple[AdmittedNodeBinding, AdmittedNodeBinding, AdmittedNodeBinding]
|
|
start_node_id: str
|
|
tool_node_id: str
|
|
end_node_id: str
|
|
|
|
|
|
def _refuse(reason: str = "unsupported_json_tool_configuration") -> NoReturn:
|
|
raise BuilderExecutionPolicyError(reason)
|
|
|
|
|
|
def _bounded_text(value: object, limit: int) -> str:
|
|
if type(value) is not str:
|
|
_refuse()
|
|
if len(value) > limit:
|
|
_refuse("json_execution_limit")
|
|
try:
|
|
size = len(value.encode("utf-8"))
|
|
except UnicodeError:
|
|
_refuse()
|
|
if size > limit:
|
|
_refuse("json_execution_limit")
|
|
return value
|
|
|
|
|
|
def _validate_path(value: object) -> str:
|
|
path = _bounded_text(value, _MAX_FILTER)
|
|
if not path.startswith("$"):
|
|
_refuse("unsupported_json_filter")
|
|
index, count, wildcards = 1, 0, 0
|
|
while index < len(path):
|
|
match = _PATH_PART.match(path, index)
|
|
if match is None:
|
|
_refuse("unsupported_json_filter")
|
|
count += 1
|
|
if count > 16:
|
|
_refuse("unsupported_json_filter")
|
|
part = match.group(0)
|
|
if part == "[*]":
|
|
wildcards += 1
|
|
if wildcards > 1:
|
|
_refuse("unsupported_json_filter")
|
|
elif match.group(2) is not None and int(match.group(2)) > 65535:
|
|
_refuse("unsupported_json_filter")
|
|
index = match.end()
|
|
return path
|
|
|
|
|
|
def _validate_content(value: object, *, constant: bool) -> str:
|
|
content = _bounded_text(value, _MAX_CONTENT)
|
|
if constant and _UNSAFE_CONTENT.search(content):
|
|
_refuse("unsupported_json_content_constant")
|
|
return content
|
|
|
|
|
|
def _start_variables(data: Mapping[str, Any]) -> set[str]:
|
|
variables = data.get("variables")
|
|
if not isinstance(variables, list) or len(variables) > 128:
|
|
_refuse("unsupported_start_input")
|
|
names: set[str] = set()
|
|
for item in variables:
|
|
if not isinstance(item, dict) or set(item) - {"variable", "type", "label", "required", "max_length", "options"}:
|
|
_refuse("unsupported_start_input")
|
|
name = item.get("variable")
|
|
if (
|
|
type(name) is not str
|
|
or not _START_NAME.fullmatch(name)
|
|
or name in names
|
|
or item.get("type") not in {"text-input", "paragraph", "select"}
|
|
or ("required" in item and type(item["required"]) is not bool)
|
|
or (
|
|
"max_length" in item
|
|
and (type(item["max_length"]) is not int or not 0 <= item["max_length"] <= _MAX_CONTENT)
|
|
)
|
|
or (
|
|
"options" in item
|
|
and (
|
|
not isinstance(item["options"], list) or any(type(option) is not str for option in item["options"])
|
|
)
|
|
)
|
|
):
|
|
_refuse("unsupported_start_input")
|
|
names.add(name)
|
|
return names
|
|
|
|
|
|
def _string_source(raw: object, *, parameter: str, start_id: str, names: set[str]) -> tuple[str, str]:
|
|
if not isinstance(raw, dict) or set(raw) != {"type", "value"}:
|
|
_refuse()
|
|
kind, value = raw["type"], raw["value"]
|
|
if kind == "constant":
|
|
if parameter == "content":
|
|
return kind, _validate_content(value, constant=True)
|
|
return kind, _validate_path(value)
|
|
if kind == "variable":
|
|
if (
|
|
not isinstance(value, list)
|
|
or len(value) != 2
|
|
or type(value[0]) is not str
|
|
or type(value[1]) is not str
|
|
or value[0] != start_id
|
|
or value[1] not in names
|
|
):
|
|
_refuse("unsupported_json_selector")
|
|
return kind, value[1]
|
|
if kind == "mixed":
|
|
if type(value) is not str:
|
|
_refuse("unsupported_json_selector")
|
|
match = _PURE_MIXED.fullmatch(value)
|
|
if match is None or match.group(1) != start_id or match.group(2) not in names:
|
|
_refuse("unsupported_json_selector")
|
|
return kind, match.group(2)
|
|
_refuse("unsupported_json_selector")
|
|
|
|
|
|
def _ensure_ascii(value: object) -> bool:
|
|
if type(value) is bool:
|
|
return value
|
|
if isinstance(value, dict) and set(value) == {"type", "value"} and value["type"] == "constant":
|
|
if type(value["value"]) is bool:
|
|
return value["value"]
|
|
_refuse()
|
|
|
|
|
|
def _preflight_tool(data: Mapping[str, Any], start_id: str, names: set[str]) -> None:
|
|
if set(data) - _TOOL_DATA or data.get("type") != "tool" or data.get("version", "1") != "1":
|
|
_refuse()
|
|
if (
|
|
data.get("tool_node_version") != "2"
|
|
or data.get("provider_type") != "builtin"
|
|
or data.get("provider_id") != _PROVIDER_ID
|
|
or data.get("plugin_id") != _PLUGIN_ID
|
|
or data.get("tool_name") != "parse"
|
|
or (
|
|
data.get("plugin_unique_identifier") is not None
|
|
and data.get("plugin_unique_identifier") != _UNIQUE_IDENTIFIER
|
|
)
|
|
or data.get("provider_name") not in (None, _PROVIDER_ID)
|
|
):
|
|
_refuse()
|
|
if any(data.get(key) is not None for key in ("credential_id", "credential_type", "credential_name")):
|
|
_refuse("unsupported_json_credentials")
|
|
# Native editor display fields are inert here; credentials and dispatch still use the qualified binding.
|
|
if "is_team_authorization" in data and type(data["is_team_authorization"]) is not bool:
|
|
_refuse()
|
|
for key, limit in (
|
|
("provider_show_name", _MAX_PROVIDER_SHOW_NAME),
|
|
("provider_icon", _MAX_PROVIDER_ICON),
|
|
("provider_icon_dark", _MAX_PROVIDER_ICON),
|
|
):
|
|
if key in data:
|
|
_bounded_text(data[key], limit)
|
|
if data.get("output_schema", {}) != {}:
|
|
_refuse()
|
|
parameters, configurations = data.get("tool_parameters"), data.get("tool_configurations")
|
|
if (
|
|
not isinstance(parameters, dict)
|
|
or set(parameters) != {"content", "json_filter"}
|
|
or not isinstance(configurations, dict)
|
|
or set(configurations) != {"ensure_ascii"}
|
|
):
|
|
_refuse()
|
|
_ensure_ascii(configurations["ensure_ascii"])
|
|
for name in ("content", "json_filter"):
|
|
_string_source(parameters[name], parameter=name, start_id=start_id, names=names)
|
|
schemas = data.get("paramSchemas")
|
|
if not isinstance(schemas, list) or len(schemas) != 3:
|
|
_refuse()
|
|
schema_by_name = {}
|
|
for schema in schemas:
|
|
if (
|
|
not isinstance(schema, dict)
|
|
or set(schema) - _PARAM_SCHEMA_KEYS
|
|
or type(schema.get("name")) is not str
|
|
or schema["name"] in schema_by_name
|
|
):
|
|
_refuse()
|
|
schema_by_name[schema["name"]] = schema
|
|
if set(schema_by_name) != {"content", "json_filter", "ensure_ascii"}:
|
|
_refuse()
|
|
for name, form, kind in (
|
|
("content", "llm", "string"),
|
|
("json_filter", "llm", "string"),
|
|
("ensure_ascii", "form", "boolean"),
|
|
):
|
|
schema = schema_by_name[name]
|
|
if (
|
|
schema.get("form") != form
|
|
or schema.get("type") != kind
|
|
or schema.get("scope") is not None
|
|
or schema.get("auto_generate") is not None
|
|
or schema.get("input_schema") is not None
|
|
or schema.get("template") is not None
|
|
or schema.get("multiple") is not False
|
|
or schema.get("options") != []
|
|
or schema.get("required") is not (name != "ensure_ascii")
|
|
or schema.get("default") != (True if name == "ensure_ascii" else None)
|
|
):
|
|
_refuse()
|
|
if data.get("params") != {"content": "", "json_filter": "", "ensure_ascii": ""}:
|
|
_refuse()
|
|
|
|
|
|
def admit_json_workflow(snapshot: RestrictedAdmissionSnapshot) -> JsonToolAdmission:
|
|
"""Require exactly the audited native Start→Tool→End graph and raw values."""
|
|
from pydantic import ValidationError
|
|
|
|
from graphon.nodes.end.end_node import EndNode
|
|
from graphon.nodes.start.start_node import StartNode
|
|
from graphon.nodes.tool.tool_node import ToolNode
|
|
from services.workflow_runtime.node_factory import resolve_workflow_node_class
|
|
|
|
admit_raw_execution_metadata(snapshot)
|
|
graph = snapshot.graph
|
|
if set(graph) - _GRAPH_KEYS:
|
|
_refuse("unsupported_graph")
|
|
nodes, edges = graph.get("nodes"), graph.get("edges")
|
|
if not isinstance(nodes, list) or len(nodes) != 3 or not isinstance(edges, list) or len(edges) != 2:
|
|
_refuse("unsupported_graph")
|
|
by_type: dict[str, tuple[str, dict[str, Any]]] = {}
|
|
seen: set[str] = set()
|
|
for node in nodes:
|
|
if not isinstance(node, dict) or set(node) - _NODE_KEYS or not isinstance(node.get("data"), dict):
|
|
_refuse("unsupported_graph")
|
|
data, node_id = node["data"], node.get("id")
|
|
if (
|
|
type(node_id) is not str
|
|
or not _NODE_ID.fullmatch(node_id)
|
|
or node_id in seen
|
|
or node_id in _RESERVED_NATIVE_OWNERS
|
|
):
|
|
_refuse("unsupported_graph")
|
|
if any(
|
|
node.get(key) or data.get(key)
|
|
for key in ("parentId", "parent_id", "iteration_id", "loop_id", "isInIteration", "isInLoop")
|
|
):
|
|
_refuse("unsupported_graph")
|
|
kind = data.get("type")
|
|
if type(kind) is not str or kind not in {"start", "tool", "end"} or kind in by_type:
|
|
_refuse("unsupported_graph")
|
|
seen.add(node_id)
|
|
by_type[kind] = (node_id, data)
|
|
if set(by_type) != {"start", "tool", "end"}:
|
|
_refuse("unsupported_graph")
|
|
start_id, start_data = by_type["start"]
|
|
tool_id, tool_data = by_type["tool"]
|
|
end_id, end_data = by_type["end"]
|
|
if set(start_data) - _START_DATA or set(end_data) - _END_DATA:
|
|
_refuse()
|
|
names = _start_variables(start_data)
|
|
_preflight_tool(tool_data, start_id, names)
|
|
outputs = end_data.get("outputs")
|
|
if not isinstance(outputs, list) or not outputs or len(outputs) > 128:
|
|
_refuse("unsupported_selector")
|
|
tool_output = False
|
|
output_names: set[str] = set()
|
|
for output in outputs:
|
|
if not isinstance(output, dict) or set(output) not in (
|
|
{"variable", "value_selector"},
|
|
{"variable", "value_selector", "value_type"},
|
|
):
|
|
_refuse("unsupported_selector")
|
|
if "value_type" in output and (type(output["value_type"]) is not str or output["value_type"] != "string"):
|
|
_refuse("unsupported_selector")
|
|
name, selector = output["variable"], output["value_selector"]
|
|
if type(name) is not str or not _START_NAME.fullmatch(name) or name in output_names:
|
|
_refuse("unsupported_selector")
|
|
output_names.add(name)
|
|
if not isinstance(selector, list) or len(selector) != 2 or any(type(x) is not str for x in selector):
|
|
_refuse("unsupported_selector")
|
|
if selector == [tool_id, "text"]:
|
|
tool_output = True
|
|
elif selector[0] != start_id or selector[1] not in names:
|
|
_refuse("unsupported_selector")
|
|
if not tool_output:
|
|
_refuse("unsupported_selector")
|
|
connections = set()
|
|
for edge in edges:
|
|
if (
|
|
not isinstance(edge, dict)
|
|
or set(edge) - _EDGE_KEYS
|
|
or edge.get("sourceHandle", "source") != "source"
|
|
or edge.get("targetHandle", "target") != "target"
|
|
):
|
|
_refuse("unsupported_graph")
|
|
if not all(type(edge.get(key)) is str for key in ("source", "target")):
|
|
_refuse("unsupported_graph")
|
|
edge_data = edge.get("data", {})
|
|
if not isinstance(edge_data, dict) or any(
|
|
edge_data.get(key) for key in ("isInIteration", "isInLoop", "iteration_id", "loop_id")
|
|
):
|
|
_refuse("unsupported_graph")
|
|
connections.add((edge["source"], edge["target"]))
|
|
if connections != {(start_id, tool_id), (tool_id, end_id)}:
|
|
_refuse("unsupported_graph")
|
|
# Source checks above run before graph hashing, native class construction or logging.
|
|
canonical_digest(graph)
|
|
audited = {"start": StartNode, "tool": ToolNode, "end": EndNode}
|
|
bindings = []
|
|
for kind in ("start", "tool", "end"):
|
|
node_id, data = by_type[kind]
|
|
version = data.get("version", "1")
|
|
if version != "1":
|
|
_refuse("unsupported_node_implementation")
|
|
cls = resolve_workflow_node_class(node_type=kind, node_version=version, node_data=data)
|
|
if cls is not audited[kind] or cls.version() != "1":
|
|
_refuse("unsupported_node_implementation")
|
|
try:
|
|
cls.validate_node_data(data)
|
|
except (ValidationError, ValueError, TypeError):
|
|
_refuse("invalid_node_configuration")
|
|
bindings.append(
|
|
AdmittedNodeBinding(
|
|
node_id=node_id,
|
|
implementation=f"{cls.__module__}.{cls.__qualname__}",
|
|
node_version="1",
|
|
normalized_config_digest=node_config_digest(data),
|
|
)
|
|
)
|
|
return JsonToolAdmission(
|
|
nodes=(bindings[0], bindings[1], bindings[2]), start_node_id=start_id, tool_node_id=tool_id, end_node_id=end_id
|
|
)
|
|
|
|
|
|
def resolve_json_parameters(
|
|
snapshot: RestrictedAdmissionSnapshot, effective_inputs: Mapping[str, object]
|
|
) -> dict[str, str | bool]:
|
|
"""Compute the expected scalar envelope solely from owned effective Start inputs."""
|
|
admitted = admit_json_workflow(snapshot)
|
|
if not isinstance(effective_inputs, Mapping):
|
|
_refuse("unsupported_input_value")
|
|
nodes = {node["id"]: node["data"] for node in snapshot.graph["nodes"]}
|
|
data = nodes[admitted.tool_node_id]
|
|
names = _start_variables(nodes[admitted.start_node_id])
|
|
resolved: dict[str, str | bool] = {}
|
|
for parameter in ("content", "json_filter"):
|
|
kind, source = _string_source(
|
|
data["tool_parameters"][parameter], parameter=parameter, start_id=admitted.start_node_id, names=names
|
|
)
|
|
if kind == "constant":
|
|
resolved[parameter] = source
|
|
else:
|
|
value = effective_inputs.get(source)
|
|
resolved[parameter] = (
|
|
_validate_content(value, constant=False) if parameter == "content" else _validate_path(value)
|
|
)
|
|
resolved["ensure_ascii"] = _ensure_ascii(data["tool_configurations"]["ensure_ascii"])
|
|
validate_json_parameters(resolved)
|
|
return resolved
|
|
|
|
|
|
def validate_json_parameters(parameters: Mapping[str, object]) -> None:
|
|
"""Validate the final native scalar envelope before its digest or dispatch."""
|
|
if not isinstance(parameters, Mapping) or set(parameters) != {"content", "json_filter", "ensure_ascii"}:
|
|
_refuse("invalid_json_parameters")
|
|
_validate_content(parameters["content"], constant=False)
|
|
_validate_path(parameters["json_filter"])
|
|
if type(parameters["ensure_ascii"]) is not bool:
|
|
_refuse("invalid_json_parameters")
|