feat(036): agent context v2 with scenario intent validation and SQL-free allowlist

This commit is contained in:
2026-07-28 08:30:41 +03:00
parent d90ad81f6d
commit 39ddfa227c
2 changed files with 51 additions and 14 deletions

View File

@@ -3,12 +3,13 @@
# @ingroup AgentChat
# @BRIEF UIContext validation and prompt-injection protection.
# @LAYER Service
# @POST Passes through contextVersion, objectType, objectId, objectName, envId, route, padding.
# @INVARIANT contextVersion must be 1 or absent (defaults to 1).
# @POST Passes through contextVersion, objectType, objectId, objectName, envId, route, intent, padding.
# @INVARIANT contextVersion must be 1 or 2. v2 scenario intent requires objectType=dashboard.
# @INVARIANT Serialized payload must not exceed 4096 bytes.
import json
ALLOWED_OBJECT_TYPES: frozenset = frozenset({"dashboard", "dataset", "migration"})
_VALID_INTENTS: frozenset = frozenset({"build_dashboard_test_scenario"})
_MAX_PAYLOAD_BYTES = 4096
_MAX_OBJECT_NAME_LENGTH = 256
_MAX_ROUTE_LENGTH = 512
@@ -75,14 +76,20 @@ def _check_route(value: str) -> None:
# #endregion AgentChat.Context.Validate.CheckRoute
# #region AgentChat.Context.Validate.CheckContextVersion [C:1] [TYPE Function] [SEMANTICS agent-chat,context,validate,version]
# #region AgentChat.Context.Validate.CheckContextVersion [C:2] [TYPE Function] [SEMANTICS agent-chat,context,validate,version,v2]
# @ingroup AgentChat
# @BRIEF Validate contextVersion is 1.
def _check_context_version(value: int | None) -> None:
# @BRIEF Validate contextVersion is 1 or 2. v2 scenario intent requires strict checks.
# @POST Raises UIContextValidationError on invalid version or invalid v2 intent.
def _check_context_version(value: int | None, intent: str | None = None, object_type: str | None = None) -> None:
if value is None:
raise UIContextValidationError("UIContext: contextVersion is required")
if value != 1:
raise UIContextValidationError(f"UIContext: unsupported contextVersion '{value}'")
if value not in (1, 2):
raise UIContextValidationError(f"UIContext: unsupported contextVersion '{value}' — must be 1 or 2")
if value == 2:
if object_type != "dashboard":
raise UIContextValidationError("UIContext v2: objectType must be 'dashboard'")
if intent is not None and intent not in _VALID_INTENTS:
raise UIContextValidationError(f"UIContext v2: unsupported intent '{intent}' — must be one of {_VALID_INTENTS}")
# #endregion AgentChat.Context.Validate.CheckContextVersion
@@ -118,10 +125,14 @@ def validate_uicontext(raw: dict) -> dict:
# payloads to prevent prompt injection via large text fields.
_check_payload_size(raw)
validated = dict(raw) # Preserve ALL input fields including contextVersion
validated = dict(raw) # Preserve ALL input fields including contextVersion and intent
# Validate known fields
_check_context_version(validated.get("contextVersion"))
# Validate known fields — detect v2 and route to strict checks
_check_context_version(
validated.get("contextVersion"),
intent=validated.get("intent"),
object_type=validated.get("objectType"),
)
_check_object_type(validated.get("objectType"))
_check_object_id(validated.get("objectId"))
_check_object_name(validated.get("objectName"))

View File

@@ -1,7 +1,7 @@
# agent/src/ss_tools/agent/_tool_filter.py
# #region AgentChat.ToolFilter [C:3] [TYPE Module] [SEMANTICS agent-chat,tools,filter,context]
# @ingroup AgentChat
# @BRIEF Context-aware tool filtering + RBAC enforcement.
# @BRIEF Context-aware tool filtering + RBAC enforcement + scenario intent gating.
# @LAYER Service
# @DATA_CONTRACT build_tool_pipeline returns a list — never mutates the input list.
# @DATA_CONTRACT enforce_tool_permission returns bool for any string input.
@@ -10,6 +10,24 @@ from typing import Any
from ss_tools.shared.logger import logger
_SCENARIO_TOOL_ALLOWLIST: frozenset = frozenset({
"superset_list_databases",
"superset_explore_database",
"search_dashboards",
"get_health_summary",
"get_task_status",
"list_environments",
"create_branch",
"commit_changes",
"deploy_dashboard",
"run_llm_validation",
"run_llm_documentation",
"show_capabilities",
})
"""Tools allowed in dashboard-testing scenario mode.
superset_execute_sql, superset_format_sql, superset_create_dataset, and any
SQL-query tools are EXCLUDED — scenario assertions use Superset-native APIs only."""
_CONTEXT_TOOL_AFFINITY: dict[str, set[str]] = {
"dashboard": {
"superset_list_databases",
@@ -58,23 +76,31 @@ _MANDATORY_TOOLS: set[str] = {"show_capabilities"}
# #region AgentChat.ToolFilter.BuildPipeline [C:3] [TYPE Function] [SEMANTICS agent-chat,tools,filter,pipeline]
# @ingroup AgentChat
# @BRIEF Apply RBAC + context-affinity filtering to a tool list, always including mandatory tools.
# @DATA_CONTRACT Input: (tools, user_role, object_type?) -> Output: filtered list (never mutates input).
# @BRIEF Apply RBAC + context-affinity filtering + scenario allowlist to a tool list.
# @DATA_CONTRACT Input: (tools, user_role, object_type?, intent?) -> Output: filtered list (never mutates input).
# @DATA_CONTRACT Mandatory tools (show_capabilities) always pass through.
def build_tool_pipeline(
tools: list[Any],
user_role: str,
object_type: str | None = None,
intent: str | None = None,
) -> list[Any]:
filtered: list[Any] = []
scenario_mode = intent == "build_dashboard_test_scenario"
for tool in tools:
name: str = tool.name
# RBAC check first
if name in _TOOL_PERMISSIONS:
allowed_roles: list[str] = _TOOL_PERMISSIONS[name]
if user_role not in allowed_roles:
logger.reason("Tool excluded by RBAC", payload={"tool": name, "reason": f"role '{user_role}' not in allowed roles {allowed_roles}"}, extra={"src": "AgentChat.ToolFilter"})
continue
if (object_type is not None and object_type in _CONTEXT_TOOL_AFFINITY and name not in _CONTEXT_TOOL_AFFINITY[object_type] and name not in _MANDATORY_TOOLS):
# Scenario allowlist check — blocks all arbitrary-SQL tools
if scenario_mode and name not in _SCENARIO_TOOL_ALLOWLIST and name not in _MANDATORY_TOOLS:
logger.reason("Tool excluded by scenario allowlist", payload={"tool": name, "reason": "not in scenario allowlist (no arbitrary SQL)"}, extra={"src": "AgentChat.ToolFilter"})
continue
# Context affinity check (only when not in scenario mode — scenario uses allowlist instead)
if not scenario_mode and (object_type is not None and object_type in _CONTEXT_TOOL_AFFINITY and name not in _CONTEXT_TOOL_AFFINITY[object_type] and name not in _MANDATORY_TOOLS):
logger.reason("Tool excluded by context", payload={"tool": name, "reason": f"not in context affinity set for object_type '{object_type}'"}, extra={"src": "AgentChat.ToolFilter"})
continue
filtered.append(tool)