"""Pure, conservative Builder effect admission; shared DTOs live in services.""" import json import math import re from collections.abc import Mapping from typing import Any from pydantic import ValidationError from runtime_contracts.builder.execution import ( MAX_OBSERVATIONS as MAX_OBSERVATIONS, # noqa: PLC0414 - preserve historical core import path ) from runtime_contracts.builder.execution import ( AdmittedNodeBinding, BuilderExecutionContext, BuilderExecutionPolicyError, _TrustedModel, canonical_digest, context_digest, fixture_digest, ) from runtime_contracts.builder.execution import ( BoundedID as BoundedID, # noqa: PLC0414 - preserve historical core import path ) from runtime_contracts.builder.execution import ( BuilderExecutionContextValue as BuilderExecutionContextValue, # noqa: PLC0414 - preserve historical core import path ) from runtime_contracts.builder.execution import ( BuilderExecutionObservation as BuilderExecutionObservation, # noqa: PLC0414 - preserve historical core import path ) from runtime_contracts.builder.execution import ( BuilderExecutionRecorder as BuilderExecutionRecorder, # noqa: PLC0414 - preserve historical core import path ) from runtime_contracts.builder.execution import ( BuilderExecutionRefusal as BuilderExecutionRefusal, # noqa: PLC0414 - preserve historical core import path ) from runtime_contracts.builder.execution import ( ConfirmedToolExecutionContext as ConfirmedToolExecutionContext, # noqa: PLC0414 - preserve historical core import path ) from runtime_contracts.builder.execution import ( Digest as Digest, # noqa: PLC0414 - preserve historical core import path ) from runtime_contracts.builder.execution import ( ExecutionEvidenceSummary as ExecutionEvidenceSummary, # noqa: PLC0414 - preserve historical core import path ) from runtime_contracts.builder.execution import ( HttpFixtureSetV1 as HttpFixtureSetV1, # noqa: PLC0414 - preserve historical core import path ) from runtime_contracts.builder.execution import ( HttpResponseFixtureV1 as HttpResponseFixtureV1, # noqa: PLC0414 - preserve historical core import path ) from runtime_contracts.builder.execution import ( JsonToolExecutionContext as JsonToolExecutionContext, # noqa: PLC0414 - preserve historical core import path ) from runtime_contracts.builder.execution import ( QualifiedJsonToolBinding as QualifiedJsonToolBinding, # noqa: PLC0414 - preserve historical core import path ) from runtime_contracts.builder.execution import ( ReasonCode as ReasonCode, # noqa: PLC0414 - preserve historical core import path ) from runtime_contracts.builder.execution import ( ScalarValue as ScalarValue, # noqa: PLC0414 - preserve historical core import path ) from runtime_contracts.builder.execution import ( TrustedExecutionCompletion as TrustedExecutionCompletion, # noqa: PLC0414 - preserve historical core import path ) from runtime_contracts.builder.execution import ( canonical_json as canonical_json, # noqa: PLC0414 - preserve historical core import path ) from runtime_contracts.builder.execution import ( decode_http_fixture_set as decode_http_fixture_set, # noqa: PLC0414 - preserve historical core import path ) from runtime_contracts.builder.execution import ( decode_http_fixture_submission as decode_http_fixture_submission, # noqa: PLC0414 - preserve historical core import path ) from runtime_contracts.builder.execution import ( scalar_inputs_digest as scalar_inputs_digest, # noqa: PLC0414 - preserve historical core import path ) from runtime_contracts.builder.execution import ( validate_execution_context as validate_execution_context, # noqa: PLC0414 - preserve historical core import path ) class RestrictedAdmissionSnapshot(_TrustedModel): """Untrusted raw graph/config snapshot, held only for inspection.""" app_mode: str workflow_kind: str graph: dict[str, Any] input_schema: dict[str, Any] features: dict[str, Any] has_environment_variables: bool has_conversation_variables: bool has_external_tracing: bool _DISABLED_FEATURES = frozenset( { "file_upload", "speech_to_text", "text_to_speech", "retriever_resource", "sensitive_word_avoidance", "suggested_questions_after_answer", "annotation_reply", "more_like_this", } ) def _admit_disabled_feature(key: str, value: Any) -> None: """Accept reviewed native disabled shapes without feature-owner side effects.""" if not isinstance(value, dict) or value.get("enabled", False) is not False: raise BuilderExecutionPolicyError("unsupported_features") string_fields: set[str] = set() list_fields: set[str] = set() allowed = {"enabled"} if key == "text_to_speech": string_fields = {"voice", "language"} allowed |= string_fields | {"autoPlay"} if "autoPlay" in value and value["autoPlay"] not in ("enabled", "disabled"): raise BuilderExecutionPolicyError("unsupported_features") elif key == "suggested_questions_after_answer": string_fields = {"prompt"} allowed |= string_fields elif key == "sensitive_word_avoidance": allowed |= {"type", "config", "configs"} if value.get("type", "") != "" or value.get("config", {}) != {} or value.get("configs", []) != []: raise BuilderExecutionPolicyError("unsupported_features") elif key == "file_upload": list_fields = {"allowed_file_types", "allowed_file_extensions", "allowed_file_upload_methods"} allowed |= list_fields | {"image", "number_limits"} if "number_limits" in value and (type(value["number_limits"]) is not int or value["number_limits"] < 0): raise BuilderExecutionPolicyError("unsupported_features") if "image" in value: image = value["image"] if ( not isinstance(image, dict) or image.get("enabled", False) is not False or set(image) - {"enabled", "number_limits", "detail", "transfer_methods"} or ("detail" in image and image["detail"] not in ("high", "low")) or ( "number_limits" in image and (type(image["number_limits"]) is not int or image["number_limits"] < 0) ) or ( "transfer_methods" in image and ( not isinstance(image["transfer_methods"], list) or any(method not in ("local_file", "remote_url") for method in image["transfer_methods"]) ) ) ): raise BuilderExecutionPolicyError("unsupported_features") if ( set(value) - allowed or any(name in value and not isinstance(value[name], str) for name in string_fields) or any( name in value and (not isinstance(value[name], list) or any(not isinstance(v, str) for v in value[name])) for name in list_fields ) ): raise BuilderExecutionPolicyError("unsupported_features") def admit_raw_execution_metadata(snapshot: RestrictedAdmissionSnapshot) -> None: from services.dify_builder.execution_features import is_citation_display_metadata if snapshot.app_mode != "workflow" or snapshot.workflow_kind != "standard": raise BuilderExecutionPolicyError("unsupported_workflow") if snapshot.has_environment_variables or snapshot.has_conversation_variables: raise BuilderExecutionPolicyError("unsupported_variables") if snapshot.has_external_tracing: raise BuilderExecutionPolicyError("unsupported_external_tracing") for key, value in snapshot.features.items(): if key == "retriever_resource" and is_citation_display_metadata(value): continue if key in {"opening_statement", "suggested_questions", "external_data_tools", "external_data_variables"}: if value not in ("", [], None): raise BuilderExecutionPolicyError("unsupported_features") elif key in _DISABLED_FEATURES: _admit_disabled_feature(key, value) else: raise BuilderExecutionPolicyError("unsupported_features") canonical_digest(snapshot.features) def validate_restricted_http_defaults(data: Mapping[str, Any]) -> None: """Only the reviewed native scalar HTTP default-output fields are admitted.""" if data.get("default_values") not in (None, []): raise BuilderExecutionPolicyError("unsupported_http_defaults") defaults = data.get("default_value") if defaults is None: return if not isinstance(defaults, list) or len(defaults) > 2: raise BuilderExecutionPolicyError("unsupported_http_defaults") seen: set[str] = set() for item in defaults: if not isinstance(item, dict) or set(item) != {"type", "key", "value"}: raise BuilderExecutionPolicyError("unsupported_http_defaults") key, kind, value = item["key"], item["type"], item["value"] if not isinstance(key, str) or key in seen: raise BuilderExecutionPolicyError("unsupported_http_defaults") seen.add(key) if key == "body" and kind == "string" and type(value) is str: try: if len(value.encode("utf-8")) <= 65536: continue except UnicodeError: pass if key == "status_code" and kind == "number" and type(value) in (int, float): try: if math.isfinite(value): continue except OverflowError: # Refuse integers outside the native finite float range before # concrete validation or any later native number conversion. pass raise BuilderExecutionPolicyError("unsupported_http_defaults") def admitted_node_bindings(snapshot: RestrictedAdmissionSnapshot) -> tuple[AdmittedNodeBinding, ...]: """Resolve audited class identities before class-owned data validation.""" from graphon.nodes.end.end_node import EndNode from graphon.nodes.http_request.node import HttpRequestNode from graphon.nodes.start.start_node import StartNode from graphon.nodes.template_transform.template_transform_node import TemplateTransformNode from services.dify_builder.revision import node_config_digest from services.dify_builder.template_execution import validate_template_configuration, validate_template_source from services.workflow_runtime.node_factory import resolve_workflow_node_class admit_raw_execution_metadata(snapshot) graph = snapshot.graph nodes, edges = graph.get("nodes"), graph.get("edges") if not isinstance(nodes, list) or not 2 <= len(nodes) <= 128 or not isinstance(edges, list): raise BuilderExecutionPolicyError("unsupported_graph") # Source limits precede the existing graph digest's UTF-8 allocation. for node in nodes: if isinstance(node, dict) and isinstance(node.get("data"), dict): if node["data"].get("type") == "template-transform": validate_template_source(node["data"].get("template")) canonical_digest(graph) audited = { "start": StartNode, "end": EndNode, "http-request": HttpRequestNode, "template-transform": TemplateTransformNode, } bindings = [] types: dict[str, str] = {} raw_data: dict[str, dict[str, Any]] = {} for node in nodes: if not isinstance(node, dict) or not isinstance(node.get("data"), dict): raise BuilderExecutionPolicyError("unsupported_graph") data, node_id = node["data"], node.get("id") if not isinstance(node_id, str) or not node_id or len(node_id) > 128 or node_id in types: raise BuilderExecutionPolicyError("unsupported_graph") if any( node.get(k) or data.get(k) for k in ("parentId", "parent_id", "iteration_id", "loop_id", "isInIteration", "isInLoop") ): raise BuilderExecutionPolicyError("unsupported_graph") node_type, version = data.get("type"), data.get("version", "1") if not isinstance(node_type, str) or node_type not in audited or version != "1": raise BuilderExecutionPolicyError("unsupported_node_implementation") cls = resolve_workflow_node_class(node_type=node_type, node_version=version, node_data=data) if cls is not audited[node_type] or cls.version() != "1": raise BuilderExecutionPolicyError("unsupported_node_implementation") if node_type == "http-request": validate_restricted_http_defaults(data) elif node_type == "template-transform": validate_template_configuration(data) try: cls.validate_node_data(data) except (ValidationError, ValueError, TypeError): raise BuilderExecutionPolicyError("invalid_node_configuration") from None bindings.append( AdmittedNodeBinding( node_id=node_id, implementation=f"{cls.__module__}.{cls.__qualname__}", node_version="1", normalized_config_digest=node_config_digest(data), ) ) types[node_id], raw_data[node_id] = node_type, data starts, ends = [n for n, t in types.items() if t == "start"], [n for n, t in types.items() if t == "end"] if len(starts) != 1 or len(ends) != 1 or len(edges) != len(nodes) - 1: raise BuilderExecutionPolicyError("unsupported_graph") outgoing: dict[str, str] = {} incoming: set[str] = set() for edge in edges: if not isinstance(edge, dict): raise BuilderExecutionPolicyError("unsupported_graph") source, target = edge.get("source"), edge.get("target") if ( not isinstance(source, str) or not isinstance(target, str) or source not in types or target not in types or source in outgoing or target in incoming or source == ends[0] or target == starts[0] ): raise BuilderExecutionPolicyError("unsupported_graph") if edge.get("sourceHandle", "source") != "source" or edge.get("targetHandle", "target") != "target": raise BuilderExecutionPolicyError("unsupported_graph") outgoing[source] = target incoming.add(target) seen: set[str] = set() current = starts[0] upstream: dict[str, set[str]] = {} while current not in seen: upstream[current] = set(seen) seen.add(current) if current not in outgoing: break current = outgoing[current] if len(seen) != len(nodes) or current != ends[0]: raise BuilderExecutionPolicyError("unsupported_graph") variables = raw_data[starts[0]].get("variables", []) if not isinstance(variables, list) or any( not isinstance(v, dict) or v.get("type") not in {"text-input", "paragraph", "number", "checkbox", "select"} for v in variables ): raise BuilderExecutionPolicyError("unsupported_start_input") start_outputs = {v.get("variable") for v in variables} def selector_allowed(node_id: str, selector: Any, *, template_input: bool = False) -> bool: if ( not isinstance(selector, (list, tuple)) or len(selector) != 2 or not isinstance(selector[0], str) or not isinstance(selector[1], str) or selector[0] not in upstream[node_id] ): return False outputs = { "start": start_outputs, "http-request": {"body"} if template_input else {"body", "status_code", "headers"}, "template-transform": {"output"}, } return selector[1] in outputs[types[selector[0]]] for node_id, data in raw_data.items(): if types[node_id] == "template-transform": if any( not selector_allowed(node_id, variable["value_selector"], template_input=True) for variable in data["variables"] ): raise BuilderExecutionPolicyError("unsupported_template_selector") if types[node_id] == "http-request": auth = data.get("authorization") if ( not isinstance(auth, dict) or auth.get("type") != "no-auth" or set(auth) - {"type", "config"} or auth.get("config") not in (None, {}) ): raise BuilderExecutionPolicyError("unsupported_http_authorization") if data.get("method", "").upper() not in {"GET", "HEAD", "POST", "PUT", "PATCH", "DELETE"}: raise BuilderExecutionPolicyError("unsupported_http_method") headers = data.get("headers", "") if headers and ( not isinstance(headers, str) or headers.lower().strip() not in { "content-type:application/json", "content-type: application/json", "content-type:text/plain", "content-type: text/plain", "content-type:application/json; charset=utf-8", "content-type: application/json; charset=utf-8", "content-type:text/plain; charset=utf-8", "content-type: text/plain; charset=utf-8", } ): raise BuilderExecutionPolicyError("unsupported_http_headers") body = data.get("body") if body is not None: if not isinstance(body, dict) or body.get("type") not in {"none", "raw-text", "json"}: raise BuilderExecutionPolicyError("unsupported_http_body") items = body.get("data", []) if isinstance(items, str): items = [{"type": "text", "value": items}] if not isinstance(items, list) or any( not isinstance(item, dict) or item.get("type") != "text" or item.get("file") for item in items ): raise BuilderExecutionPolicyError("unsupported_http_body") validate_restricted_http_defaults(data) if types[node_id] == "end": if any(not selector_allowed(node_id, output.get("value_selector")) for output in data.get("outputs", [])): raise BuilderExecutionPolicyError("unsupported_selector") serialized = json.dumps(data, ensure_ascii=False) for node, output in re.findall(r"\{\{#([^.#]+)\.([^#]+)#\}\}", serialized): if not selector_allowed(node_id, [node, output]): raise BuilderExecutionPolicyError("unsupported_selector") return tuple(bindings) def admit_restricted_workflow(context: BuilderExecutionContext, snapshot: RestrictedAdmissionSnapshot) -> None: if not isinstance(context, BuilderExecutionContext): raise BuilderExecutionPolicyError("unsupported_execution_policy") if context.context_digest != context_digest(context): raise BuilderExecutionPolicyError("context_digest_mismatch") if context.admitted_nodes != admitted_node_bindings(snapshot): raise BuilderExecutionPolicyError("node_binding_mismatch") nodes = {binding.node_id: binding for binding in context.admitted_nodes} if len({f.node_id for f in context.http_fixtures}) != len(context.http_fixtures): raise BuilderExecutionPolicyError("invalid_http_fixtures") for fixture in context.http_fixtures: binding = nodes.get(fixture.node_id) if binding is None or (binding.implementation, binding.node_version) != ( fixture.implementation, fixture.node_version, ): raise BuilderExecutionPolicyError("unused_http_fixture") if context.fixture_digest != fixture_digest(context.execution_revision, context.http_fixtures): raise BuilderExecutionPolicyError("fixture_digest_mismatch") if context.mode != ("mock" if context.http_fixtures else "restricted"): raise BuilderExecutionPolicyError("execution_mode_mismatch") if context.sandbox_profile != "disabled": raise BuilderExecutionPolicyError("sandbox_capability_unverified")